lifecyclelogging.logging

lifecyclelogging.logging

Core logging functionality for flexible and configurable logging management.

This module provides a Logging class that supports advanced logging features including:

• Configurable console and file logging
• Message storage and filtering
• Verbosity control
• Context and storage marker systems
• Clean exit with formatted output (exit_run)

The module allows for fine-grained control over log message handling, storage, and output across different logging contexts.

Module Contents

Classes

Logging	A class to manage logging configurations for console and file outputs.

Data

KeyTransform

API

lifecyclelogging.logging.KeyTransform

None

exception lifecyclelogging.logging.ExitRunError

Bases: Exception

Raised when exit_run encounters a formatting or data error.

Initialization

Initialize self. See help(type(self)) for accurate signature.

class lifecyclelogging.logging.Logging(enable_console: bool = False, enable_file: bool = True, logger: logging.Logger | None = None, logger_name: str | None = None, log_file_name: str | None = None, default_storage_marker: str | None = None, allowed_levels: collections.abc.Sequence[str] | None = None, denied_levels: collections.abc.Sequence[str] | None = None, enable_verbose_output: bool = False, verbosity_threshold: int = VERBOSITY)

A class to manage logging configurations for console and file outputs.

This class supports two types of message markers:

1. Storage markers (log_marker): Used to categorize and store messages in collections
2. Context markers (context_marker): Prepended to messages and can override verbosity

The context marker system can also designate certain markers as “verbosity bypass markers” which will cause messages with those markers to ignore verbosity settings entirely.

Initialization

Initialize the Logging class with options for console and file logging.

This class provides two types of message marking systems:

1. Storage markers: Used to categorize and collect messages in storage
2. Context markers: Used to prefix messages and control verbosity

Args: enable_console: Whether to enable console output. enable_file: Whether to enable file output. logger: An existing logger instance to use. logger_name: The name for a new logger instance. log_file_name: The name of the log file if file logging enabled. default_storage_marker: Default marker for storing messages. allowed_levels: List of allowed log levels (if empty, all allowed). denied_levels: List of denied log levels. enable_verbose_output: Whether to allow verbose messages. verbosity_threshold: Maximum verbosity level (1-5) to display.

The logger configured will have the following characteristics:

• Non-propagating (won’t pass messages to parent loggers)
• Level set from LOG_LEVEL env var or DEBUG if not set
• Console/file output based on parameters and env vars
• Gunicorn logger integration if available

static _normalize_levels(levels: collections.abc.Sequence[str] | None) → tuple[str, …]

Normalize provided log levels to lower-case tuples.

register_verbosity_bypass_marker(marker: str) → None

Add a context marker that bypasses verbosity restrictions.

_configure_logger(logger: logging.Logger | None = None, logger_name: str | None = None, log_file_name: str | None = None) → logging.Logger

Configure the logger instance.

Args: logger: An existing logger instance to use. logger_name: The name for a new logger instance. log_file_name: The name of the log file if file logging enabled.

Returns: logging.Logger: The configured logger instance.

_setup_handlers(logger: logging.Logger, log_file_name: str) → None

Set up console and file handlers.

Args: logger: The logger to which handlers will be added. log_file_name: The name of the log file for file handler.

verbosity_exceeded(verbose: bool, verbosity: int) → bool

Determines if a message should be suppressed based on verbosity settings.

Args: verbose: Flag indicating if this is a verbose message. verbosity: The verbosity level of the message (1-5).

Returns: bool: True if the message should be suppressed, False if it should be shown.

A message is not suppressed if:

1. The current context marker is in verbosity_bypass_markers
2. Verbosity level <= threshold and either:

• verbose=False, or
• verbose=True and verbose output is enabled

_prepare_message(msg: str, context_marker: str | None, identifiers: collections.abc.Sequence[str] | None) → str

Prepare the log message with context markers and identifiers.

Args: msg: The base message to prepare. context_marker: Optional marker to prefix message with and set as current context. identifiers: Optional identifiers to append in parentheses.

Returns: str: The prepared message with any context marker prefix and identifiers.

_store_logged_message(msg: str, log_level: lifecyclelogging.log_types.LogLevel, storage_marker: str | None, allowed_levels: tuple[str, …], denied_levels: tuple[str, …]) → None

Store the logged message if it meets the filtering criteria.

Args: msg: The message to store. log_level: The level the message was logged at. storage_marker: The marker to store the message under. allowed_levels: Normalized levels that are allowed (if empty, all allowed). denied_levels: Normalized levels that are denied.

Messages are stored in self.stored_messages under their storage_marker if:

1. A storage_marker is provided
2. The log_level is in allowed_levels (or allowed_levels is empty)
3. The log_level is not in denied_levels

Warning-level and above messages are prefixed with ‘:warning:’.

logged_statement(msg: str, json_data: collections.abc.Mapping[str, Any] | collections.abc.Sequence[collections.abc.Mapping[str, Any]] | None = None, labeled_json_data: collections.abc.Mapping[str, collections.abc.Mapping[str, Any]] | None = None, identifiers: collections.abc.Sequence[str] | None = None, verbose: bool = False, verbosity: int = 1, context_marker: str | None = None, log_level: lifecyclelogging.log_types.LogLevel = ‘debug’, storage_marker: str | None = None, allowed_levels: collections.abc.Sequence[str] | None = None, denied_levels: collections.abc.Sequence[str] | None = None) → str | None

Log a statement with optional data, context marking, and storage.

Args: msg: The message to log. json_data: Optional JSON data to append. labeled_json_data: Optional labeled JSON data to append. identifiers: Optional identifiers to append in parentheses. verbose: Whether this is a verbose message. verbosity: Verbosity level (1-5). context_marker: Marker to prefix message with and check for verbosity bypass. log_level: Level to log at. storage_marker: Marker for storing in message collections. allowed_levels: Override of allowed log levels. denied_levels: Override of denied log levels.

Returns: str | None: The final message if logged, None if suppressed by verbosity.

log_results(results: Any, log_file_name: str, no_formatting: bool = False, ext: str | None = None, verbose: bool = False, verbosity: int = 0) → None

Log results to a file.

Args: results: The results to log. log_file_name: Base name for the log file. no_formatting: If True, write results as-is without JSON formatting. ext: File extension (defaults to “.json”). verbose: Whether this is a verbose log. verbosity: Verbosity level for this log.

KEY_TRANSFORMS : ClassVar[dict[str, lifecyclelogging.logging.KeyTransform]]

None

_resolve_key_transform(key_transform: lifecyclelogging.logging.KeyTransform | str | None, unhump_results: bool, prefix: str | None) → lifecyclelogging.logging.KeyTransform | None

Resolve key_transform parameter to a callable.

Args: key_transform: User-provided transform (callable, string name, or None). unhump_results: Legacy flag for snake_case transformation. prefix: If set, implies transformation is needed.

Returns: A callable transform function or None.

_transform_nested_keys(data: collections.abc.Mapping[str, Any], transform_fn: lifecyclelogging.logging.KeyTransform) → dict[str, Any]

Recursively transform all keys in a nested mapping.

Args: data: The mapping to transform. transform_fn: Function to apply to each key.

Returns: A new dict with all keys transformed.

exit_run(results: collections.abc.Mapping[str, Any] | None = None, unhump_results: bool = False, key_transform: lifecyclelogging.logging.KeyTransform | str | None = None, prefix: str | None = None, prefix_allowlist: collections.abc.Sequence[str] | None = None, prefix_denylist: collections.abc.Sequence[str] | None = None, prefix_delimiter: str = ’_’, sort_by_field: str | None = None, format_results: bool = True, encode_to_base64: bool = False, encode_all_values_to_base64: bool = False, key: str | None = None, exit_on_completion: bool = True, **format_opts: Any) → Any

Format results and optionally exit the program cleanly.

This method handles the lifecycle of formatting and outputting results, typically used at the end of a data processing run. It supports:

• Error aggregation and reporting
• Result transformation (key transforms, prefixing, sorting)
• Base64 encoding
• JSON serialization
• Clean stdout output and exit

Args: results: The results to format and output. Defaults to empty dict. unhump_results: Convert camelCase keys to snake_case (shorthand for key_transform=”snake_case”). key_transform: Transform function for result keys. Can be:

• A callable that takes a string and returns a string
• A string naming a built-in transform: “snake_case”, “camel_case”, “pascal_case”, “kebab_case”
• None to skip transformation When unhump_results=True, defaults to “snake_case”. prefix: Prefix to add to result keys (implies key transformation). prefix_allowlist: Keys to include when prefixing. prefix_denylist: Keys to exclude when prefixing. prefix_delimiter: Delimiter between prefix and key (default “_”). sort_by_field: Sort results by this field’s value. format_results: Whether to format results before base64 encoding. encode_to_base64: Encode entire result to base64. encode_all_values_to_base64: Encode each top-level value to base64. key: Wrap results in a dict with this key. exit_on_completion: If True, write to stdout and exit(0). If False, return the formatted results. **format_opts: Additional options for wrap_raw_data_for_export.

Returns: If exit_on_completion=False, returns the formatted results. Otherwise, writes to stdout and exits with code 0.

Raises: RuntimeError: If there are accumulated errors in error_list. ExitRunError: If result formatting fails.

Examples:

Simple snake_case transformation (most common)

logging.exit_run(results, unhump_results=True)

# Explicit transform
logging.exit_run(results, key_transform="kebab_case")

# Custom transform function
logging.exit_run(results, key_transform=lambda k: k.upper())