Writer

Writer utilities for logging and instrumentation utilities for tracking ai model interactions.

View module source on GitHub

Module for handling log writing operations in the Maxim SDK.

This module provides classes for configuring and managing log writing, including automatic flushing, file-based persistence, and API integration.

LogWriterConfig

class LogWriterConfig()

Configuration class for the LogWriter.

Attributes:

Name	Description
base_url	Base URL for the Maxim API.
api_key	API key for authentication.
repository_id	ID of the repository to write logs to.
auto_flush	Whether to automatically flush logs periodically.
flush_interval	Time interval in seconds between automatic flushes.
is_debug	Whether to enable debug logging.
raise_exceptions	Whether to raise exceptions or handle them silently.

__init__

def __init__(base_url,
             api_key,
             repository_id,
             auto_flush=True,
             flush_interval: Optional[int] = 10,
             is_debug=False,
             raise_exceptions=False)

Initialize a LogWriterConfig instance.

Arguments:

Name	Description
base_url	Base URL for the Maxim API.
api_key	API key for authentication.
repository_id	ID of the repository to write logs to.
auto_flush	Whether to automatically flush logs periodically.
flush_interval	Time interval in seconds between automatic flushes.
is_debug	Whether to enable debug logging.
raise_exceptions	Whether to raise exceptions or handle them silently.

LogWriter

class LogWriter()

Handles writing logs to the Maxim API and local filesystem.

This class manages a queue of logs, periodically flushes them to the API, and provides fallback to local filesystem storage when API calls fail.

__init__

def __init__(config: LogWriterConfig)

Initialize a LogWriter instance.

Arguments:

Name	Description
config	Configuration for the LogWriter.

Raises:

• ValueError - If auto_flush is enabled but flush_interval is None.

repository_id

@property
def repository_id()

Get the repository ID.

Returns:

Name	Description
str	The repository ID.

upload_file_data

def upload_file_data(add_attachment_log: CommitLog)

Upload a file attachment to the Maxim API.

upload_file

def upload_file(add_attachment_log: CommitLog)

Upload a file data attachment to the Maxim API.

upload_attachments

def upload_attachments(attachment_logs: list[CommitLog])

Upload all attachments to the Maxim API.

upload_attachment

def upload_attachment(add_attachment_log: CommitLog)

Upload an attachment to the Maxim API.

Arguments:

Name	Description
attachment	Attachment object to upload.

is_running_on_lambda

def is_running_on_lambda()

Check if the code is running in an AWS Lambda environment.

Returns:

Name	Description
bool	True if running in AWS Lambda, False otherwise.

write_to_file

def write_to_file(logs)

Write logs to a local file.

Arguments:

Name	Description
logs	List of CommitLog objects to write.

Returns:

str or None: Path to the file if successful, None otherwise.

Raises:

• Exception - If raise_exceptions is True and writing fails.

flush_log_files

def flush_log_files()

Flush logs from files to the Maxim API.

This method reads log files from the logs directory, sends them to the API, and deletes the files if successful.

Raises:

• Exception - If raise_exceptions is True and an error occurs.

can_access_filesystem

def can_access_filesystem()

Check if the filesystem is accessible for writing.

Returns:

Name	Description
bool	True if filesystem is accessible, False otherwise.

flush_logs

def flush_logs(logs: list[CommitLog])

Flush logs to the Maxim API.

This method attempts to send logs to the API, with fallback mechanisms for handling failures based on the environment.

Arguments:

Name	Description
logs	List of CommitLog objects to flush.

commit

def commit(log: CommitLog)

Add a log to the queue for later flushing.

Arguments:

Name	Description
log	CommitLog object to add to the queue.

Raises:

• ValueError - If the entity_id is invalid and raise_exceptions is True.

flush_upload_attachment_logs

def flush_upload_attachment_logs(is_sync=False)

Flush all queued attachments to the Maxim API.

This method empties the queue and sends all logs to the API, with special handling for AWS Lambda environments.

flush_commit_logs

def flush_commit_logs(is_sync=False)

Flush all queued commit logs to the Maxim API.

This method empties the queue and sends all logs to the API, with special handling for AWS Lambda environments.

flush

def flush(is_sync=False)

Flush all queued logs to the Maxim API.

cleanup

def cleanup(is_sync=False)

Clean up resources used by the LogWriter.

This method stops the flush thread, flushes any remaining logs, and shuts down the executor.