Skip to content

Platform Utilities

Common utilities (logging, filesystem helpers, activity heartbeats) shared across Python services.

python_core.utils.logger

Unified logging; see SmrtHub.Logging README for rotation, schemas, and config.

Activity 

Activity(
    name: str,
    kind: str = "Internal",
    tags: Optional[dict] = None,
)

Context manager representing a tracing activity span.

:param name: Activity name recorded in structured logs. :param kind: Activity kind (defaults to "Internal"). :param tags: Optional dictionary of stringifiable tags.

add_baggage 

add_baggage(key: str, value: Optional[str])

Attach baggage propagated to downstream activities.

:param key: Baggage key. :param value: String value stored when provided. :returns: self for chaining.

add_tag 

add_tag(key: str, value)

Attach a tag to the activity.

:param key: Tag name. :param value: Value converted to string when not None. :returns: self for chaining.

ComponentLogger 

ComponentLogger(component_name: str)

Lightweight wrapper exposing logging API bound to a component.

:param component_name: Human-friendly component identifier.

log_debug 

log_debug(message: str)

Log a debug-level message for this component.

:param message: Text rendered to sinks. :returns: None.

log_error 

log_error(message: str)

Log an error-level message for this component.

:param message: Text rendered to sinks. :returns: None.

log_exception 

log_exception(message: str)

Log an exception with stack trace for this component.

:param message: Text rendered to sinks. :returns: None.

log_fatal 

log_fatal(message: str)

Log a fatal-level message for this component.

:param message: Text rendered to sinks. :returns: None.

log_info 

log_info(message: str)

Log an information-level message for this component.

:param message: Text rendered to sinks. :returns: None.

log_json 

log_json(message: str, **fields)

Emit structured-only information record with additional fields.

:param message: Rendered message text. :param fields: Additional structured properties. :returns: None.

log_warning 

log_warning(message: str)

Log a warning-level message for this component.

:param message: Text rendered to sinks. :returns: None.

SafeRotatingFileHandler 

SafeRotatingFileHandler(*args, **kwargs)

Bases: RotatingFileHandler

Rotating handler with delayed open and permission resilience (see README).

Set delay=True by default so first emit opens the file safely.

doRollover 

doRollover()

Attempt rollover; skip gracefully when the OS reports PermissionError.

begin_activity 

begin_activity(
    name: str,
    kind: str = "Internal",
    tags: Optional[dict] = None,
) -> Activity

Begin a tracing activity when trace context is enabled.

:param name: Activity name. :param kind: Activity kind label. :param tags: Optional dictionary of tags. :returns: Activity context manager.

get_correlation_id 

get_correlation_id() -> str

Return the effective correlation identifier for the current context.

:returns: Correlation identifier or empty string when unset.

get_heartbeat_logger 

get_heartbeat_logger()

Return a specialized heartbeat logger writing to the primary text log file.

:returns: logging.Logger instance emitting heartbeat messages.

get_logger_for 

get_logger_for(component: str) -> ComponentLogger

Return a logger bound to per-component files.

:param component: Component identifier. :returns: ComponentLogger writing to <slug>-log.txt and <slug>-log.json.

log_debug 

log_debug(message: str, *, component: Optional[str] = None)

Log a debug-level message with structured mirror output.

:param message: Text rendered to sinks. :param component: Optional component override for structured payloads. :returns: None.

log_error 

log_error(message: str, *, component: Optional[str] = None)

Log an error-level message with structured mirror output.

:param message: Text rendered to sinks. :param component: Optional component override for structured payloads. :returns: None.

log_exception 

log_exception(
    message: str, *, component: Optional[str] = None
)

Log an exception with stack trace mirrored into structured output.

:param message: Text rendered to sinks. :param component: Optional component override for structured payloads. :returns: None.

log_fatal 

log_fatal(message: str, *, component: Optional[str] = None)

Log a fatal-level message with structured mirror output.

:param message: Text rendered to sinks. :param component: Optional component override for structured payloads. :returns: None.

log_info 

log_info(message: str, *, component: Optional[str] = None)

Log an information-level message with structured mirror output.

:param message: Text rendered to sinks. :param component: Optional component override for structured payloads. :returns: None.

log_json 

log_json(message: str, **fields)

Emit a structured-only information record with additional fields.

:param message: Rendered message text. :param fields: Additional structured properties. :returns: None.

log_warning 

log_warning(
    message: str, *, component: Optional[str] = None
)

Log a warning-level message with structured mirror output.

:param message: Text rendered to sinks. :param component: Optional component override for structured payloads. :returns: None.

set_correlation_id 

set_correlation_id(corr_id: Optional[str])

Set the correlation identifier for the current context.

:param corr_id: Correlation identifier string or None to clear. :returns: None.

python_core.utils.config_path_resolver

Config/log path helpers; see utils README for canonical layout and overrides.

component_dir 

component_dir(
    component: str,
    *,
    roaming: bool = True,
    create: bool = True
) -> Path

Return the canonical component directory SmrtHub/Config/<slug>.

:param component: Component identifier used to derive the slug. :param roaming: True for roaming roots, False for local roots. :param create: When True the directory is created if missing. :returns: Path object pointing to the component directory.

config_file_path 

config_file_path(
    component: str,
    *,
    roaming: bool = True,
    json_ext: str = _DEF_EXT
) -> Path

Return the canonical config file path for the given component.

:param component: Component identifier. :param roaming: True for roaming roots, False for local roots. :param json_ext: Extension to use for the config file. :returns: Path pointing to <slug>-config.<ext>.

get_config_file_path 

get_config_file_path(
    component: str, *, roaming: bool = True
) -> Path

Return the canonical config file path for component.

:param component: Component identifier. :param roaming: True for roaming roots, False otherwise. :returns: Path pointing to the config file.

get_file_path 

get_file_path(
    component: str,
    kind: Literal["config", "state", "custom"],
    *,
    roaming: bool = True,
    custom_file_name: Optional[str] = None,
    json_ext: str = _DEF_EXT
) -> Path

Return a config, state, or custom file path using canonical helpers.

:param component: Component identifier. :param kind: One of "config", "state", or "custom". :param roaming: True for roaming roots, False for local roots. :param custom_file_name: File name used when kind is "custom". :param json_ext: Extension applied to config/state files. :returns: Path pointing to the requested file. :raises ValueError: If kind is unsupported or the custom name is invalid.

get_state_file_path 

get_state_file_path(
    component: str, *, roaming: bool = True
) -> Path

Return the canonical state file path for component.

:param component: Component identifier. :param roaming: True for roaming roots, False otherwise. :returns: Path pointing to the state file.

health_marker_path 

health_marker_path(
    component: str, *, roaming: bool = False
) -> Path

Return the health marker file path for the component.

:param component: Component identifier. :param roaming: True for roaming roots, False otherwise. :returns: Path for the health marker JSON file.

log_file_path 

log_file_path(
    component: str, *, roaming: bool = False
) -> Path

Return the text log file path for the component.

:param component: Component identifier. :param roaming: True to place logs under roaming roots (rare), False for local. :returns: Path pointing to <slug>-log.txt.

log_paths 

log_paths(
    component: str, *, roaming: bool = False
) -> Tuple[Path, Path]

Return both text and structured log paths for the component.

:param component: Component identifier. :param roaming: True for roaming roots, False for local roots. :returns: Tuple of (text_log_path, json_log_path).

normalize_component_slug 

normalize_component_slug(name: str) -> str

Return the canonical kebab-case slug for a component name.

Exposed for helpers that must align with Smrt.Config slug formatting rules.

:param name: Component identifier. :returns: Sanitized kebab-case slug string.

path_for 

path_for(
    component: str,
    filename: str,
    *,
    roaming: bool = True,
    create: bool = True
) -> Path

Return a path inside the component directory for an arbitrary filename.

:param component: Component identifier. :param filename: Custom filename located within the component directory. :param roaming: True for roaming roots, False otherwise. :param create: When True ensure the directory exists. :returns: Path joining the component directory and validated filename. :raises ValueError: If filename fails validation.

read_text 

read_text(
    path: Union[str, Path],
    *,
    encoding: str = "utf-8",
    default: Optional[str] = None
) -> Optional[str]

Read text from path if it exists.

:param path: File path to read. :param encoding: Text encoding of the file. :param default: Value returned when the file is missing. :returns: File contents or default when absent.

smrt_db_component_dir 

smrt_db_component_dir(
    component: str, *, create: bool = True
) -> Path

Return a component-scoped directory under the SmrtDB root.

:param component: Component identifier used to derive the slug. :param create: When True ensure the directory exists. :returns: Path pointing to SmrtDB/<slug>.

smrt_db_dir 

smrt_db_dir(*, create: bool = True) -> Path

Return the local SmrtDB directory.

:param create: When True ensure the directory exists. :returns: Path pointing to SmrtDB under the local root.

smrt_db_file_path 

smrt_db_file_path(
    component: str, name: str, *, create_dir: bool = True
) -> Path

Return a file path under a component SmrtDB directory.

:param component: Component identifier used to derive the slug directory. :param name: File name to place within the component SmrtDB directory. :param create_dir: When True ensure the directory exists. :returns: Path pointing to the requested SmrtDB entry. :raises ValueError: If name fails validation.

smrt_db_path 

smrt_db_path(name: str, *, create_dir: bool = True) -> Path

Return a file path under the SmrtDB directory for a given file name.

:param name: File name to place in the SmrtDB directory. :param create_dir: When True ensure the directory exists. :returns: Path pointing to the requested entry within SmrtDB. :raises ValueError: If name fails validation.

state_file_path 

state_file_path(
    component: str,
    *,
    roaming: bool = True,
    json_ext: str = _DEF_EXT
) -> Path

Return the canonical state file path for the given component.

:param component: Component identifier. :param roaming: True for roaming roots, False for local roots. :param json_ext: Extension to use for the state file. :returns: Path pointing to <slug>-state.<ext>.

structured_log_file_path 

structured_log_file_path(
    component: str, *, roaming: bool = False
) -> Path

Return the structured JSON log file path for the component.

:param component: Component identifier. :param roaming: True to place logs under roaming roots, False for local. :returns: Path pointing to <slug>-log.json.

validate_custom_filename 

validate_custom_filename(filename: str) -> str

Expose canonical file-name validation for shared helpers.

write_atomic 

write_atomic(
    path: Union[str, Path],
    data: Union[str, bytes, bytearray, memoryview],
    *,
    encoding: str = "utf-8"
) -> Path

Write data to path atomically via temp file replacement.

:param path: Destination path or string. :param data: Text or binary payload to persist. :param encoding: Encoding used when data is text. :returns: Path representing the final output file.

write_health_marker 

write_health_marker(
    component: str,
    status: str = "ok",
    *,
    roaming: bool = False,
    extra: Optional[dict] = None
) -> Path

Write the structured health marker JSON to the component directory.

:param component: Component identifier. :param status: Health status label stored in the marker. :param roaming: True for roaming roots, False for local roots. :param extra: Optional dictionary merged into the payload. :returns: Path of the written health marker file.

python_core.utils.filesystem_identity

Cross-platform file identity helpers; README covers platform specifics and caveats.

BY_HANDLE_FILE_INFORMATION

Bases: Structure

ctypes mirror of BY_HANDLE_FILE_INFORMATION for Windows APIs.

get_unique_file_id 

get_unique_file_id(path: Path) -> Optional[Tuple[int, int]]

Return device identifier and inode from stat on Unix-like systems.

:param path: Path pointing to a file or directory. :returns: Tuple of (device_id, inode) or None on failure.

log_debug 

log_debug(message: str)

Proxy debug logging through the shared python-core logger.

:param message: Text to record at debug level. :returns: None.

python_core.utils.fs

Safe file write utilities; README covers duplicate detection and logging expectations.

copy_file 

copy_file(source: Union[str, Path], destination: Path)

Copy file content with duplicate detection, preserving metadata via shutil.copy2.

log_debug 

log_debug(message: str)

Proxy debug logging through the shared python-core logger.

:param message: Text recorded at debug level. :returns: None.

log_error 

log_error(message: str)

Proxy error logging through the shared python-core logger.

:param message: Text recorded at error level. :returns: None.

log_info 

log_info(message: str)

Proxy info logging through the shared python-core logger.

:param message: Text recorded at info level. :returns: None.

write_file 

write_file(
    filepath: Path, content: Union[str, bytes, bytearray]
)

Write text or binary payloads after performing a duplicate-content check.

python_core.utils.heartbeat

Heartbeat thread helpers; README covers logging cadence and health markers.

Heartbeat 

Heartbeat(interval: int = 10, tag: str = 'main')

Daemon heartbeat loop.

:param interval: Interval in seconds between heartbeat ticks. :param tag: Label included in heartbeat log lines and health markers.

Validate interval, capture tag, and prepare the daemon thread.

start 

start()

Launch the daemon heartbeat loop if not already running.

stop 

stop()

Signal the heartbeat loop to exit on the next cycle.

start_heartbeat 

start_heartbeat(interval: int = 10, tag: str = 'main')

Start the singleton heartbeat loop when not already active.

:param interval: Interval in seconds between heartbeat ticks. :param tag: Label included in heartbeat log lines and health markers. :returns: None.