Logging Utils#

class ccat_data_transfer.logging_utils.BBCPLogHandler(base_log_path: str = '/var/log/ccat/bbcp_transfers')[source]#

Bases: object

__init__(base_log_path: str = '/var/log/ccat/bbcp_transfers')[source]#

get_log_path(transfer_id: int, timestamp: datetime | None = None) → Path[source]#: Generate structured log path for BBCP transfer.

store_bbcp_output(session: Session, data_transfer: DataTransfer, stdout: bytes, stderr: bytes, success: bool, timestamp: datetime | None = None) → DataTransferLog[source]#: Store BBCP output and create minimal log entry.

class ccat_data_transfer.logging_utils.StructuredLogger(logger: Logger)[source]#

Bases: object

Wrapper for structured logging with consistent formatting

__init__(logger: Logger)[source]#

setLevel(level: int) → None[source]#: Set the logging level of the logger.

getEffectiveLevel() → int[source]#: Get the effective logging level of the logger.

debug(message: str, **kwargs) → None[source]#: Log a debug message.

info(message: str, **kwargs) → None[source]#: Log an info message.

warning(message: str, **kwargs) → None[source]#: Log a warning message.

error(message: str, error: Exception | None = None, **kwargs) → None[source]#: Log an error message.

exception(message: str, **kwargs) → None[source]#: Log an exception with traceback at ERROR level. Should only be called from an except block.

ccat_data_transfer.logging_utils.get_structured_logger(name: str) → StructuredLogger[source]#: Get a structured logger instance

ccat_data_transfer.logging_utils.setup_celery_logging()[source]#: Configure Celery logging to use the same format as our application logging

Overview#

Logging configuration and utilities for the data transfer system.

Key Components#

Logging configuration
Log formatting and handlers
Logging utilities and helpers