nat.observability.mixin.file_mixin#

Attributes#

logger

Classes#

FileExportMixin

Mixin for file-based exporters.

Module Contents#

logger#

class FileExportMixin(

*args,

output_path,

project,

mode: nat.observability.mixin.file_mode.FileMode = FileMode.APPEND,

enable_rolling: bool = False,

max_file_size: int = 10 * 1024 * 1024,

max_files: int = 5,

cleanup_on_init: bool = False,

**kwargs,

)#

Bases: nat.observability.mixin.resource_conflict_mixin.ResourceConflictMixin

Mixin for file-based exporters.

This mixin provides file I/O functionality for exporters that need to write serialized data to local files, with support for file overwriting and rolling logs.

Automatically detects and prevents file path conflicts between multiple instances by raising ResourceConflictError during initialization.

Initialize the file exporter with the specified output_path and project.

Args:

output_path (str): The path to the output file or directory (if rolling enabled). project (str): The project name for metadata. mode (str): Either “append” or “overwrite”. Defaults to “append”. enable_rolling (bool): Enable rolling log files. Defaults to False. max_file_size (int): Maximum file size in bytes before rolling. Defaults to 10MB. max_files (int): Maximum number of rolled files to keep. Defaults to 5. cleanup_on_init (bool): Clean up old files during initialization. Defaults to False.

Raises:

ResourceConflictError: If another FileExportMixin instance is already using: the same file path or would create conflicting files.

_filepath#

_project#

_mode#

_enable_rolling = False#

_max_file_size = 10485760#

_max_files = 5#

_cleanup_on_init = False#

_lock#

_first_write = True#

_setup_file_paths()#: Setup file paths using the project name.

_get_resource_identifiers() → dict[str, Any]#

Return the file resources this instance will use.

Returns:: dict with file_path and optionally cleanup_pattern for rolling files.

_format_conflict_error( resource_type: str, identifier: Any, existing_instance: Any, ) → str#: Format user-friendly error messages for file conflicts.

_cleanup_old_files_sync() → None#: Synchronous version of cleanup for use during initialization.

async _should_roll_file() → bool#: Check if the current file should be rolled based on size.

async _roll_file() → None#: Roll the current file by renaming it with a timestamp and cleaning up old files.

async _cleanup_old_files() → None#: Remove old rolled files beyond the maximum count.

async export_processed(item: str | list[str]) → None#

Export a processed string or list of strings.

Args:: item (str | list[str]): The string or list of strings to export.

get_current_file_path() → pathlib.Path#

Get the current file path being written to.

Returns:: Path: The current file path being written to.

get_file_info() → dict#

Get information about the current file and rolling configuration.

Returns:

dict: A dictionary containing the current file path, mode, rolling enabled, cleanup on init,: effective project name, and additional rolling configuration if enabled.