Processors

🤖 AI-Generated Content

This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.

provide.foundation.logger.setup.processors

Classes

Functions

apply_structlog_configuration

apply_structlog_configuration(
    processors: list[Any],
    log_stream: TextIO,
    effective_level: int = 20,
) -> None

Apply the processor configuration to structlog.

Uses structlog.make_filtering_bound_logger so that methods below effective_level are literal return None — zero overhead, no processor entry, no f-string evaluation needed.

Parameters:

Name	Type	Description	Default
processors	list[Any]	List of processors to configure	required
log_stream	TextIO	Output stream for logging	required
effective_level	int	Numeric log level threshold (default 20 / INFO)	20

Source code in provide/foundation/logger/setup/processors.py

def apply_structlog_configuration(
    processors: list[Any], log_stream: TextIO, effective_level: int = 20
) -> None:
    """Apply the processor configuration to structlog.

    Uses ``structlog.make_filtering_bound_logger`` so that methods below
    *effective_level* are literal ``return None`` — zero overhead, no
    processor entry, no f-string evaluation needed.

    Args:
        processors: List of processors to configure
        log_stream: Output stream for logging
        effective_level: Numeric log level threshold (default 20 / INFO)

    """
    # Check if force stream redirect is enabled (for testing)
    # Disable caching to allow stream redirection to work properly
    from provide.foundation.streams.config import get_stream_config

    stream_config = get_stream_config()
    cache_loggers = not stream_config.force_stream_redirect

    structlog.configure(
        processors=processors,
        logger_factory=structlog.PrintLoggerFactory(file=log_stream),
        wrapper_class=_make_filtering_bound_logger_with_trace(effective_level),
        cache_logger_on_first_use=cache_loggers,
    )

build_complete_processor_chain

build_complete_processor_chain(
    config: TelemetryConfig, log_stream: TextIO
) -> list[Any]

Build the complete processor chain for structlog.

Parameters:

Name	Type	Description	Default
config	TelemetryConfig	Telemetry configuration	required
log_stream	TextIO	Output stream for logging	required

Returns:

Type	Description
list[Any]	List of processors for structlog

Source code in provide/foundation/logger/setup/processors.py

def build_complete_processor_chain(
    config: TelemetryConfig,
    log_stream: TextIO,
) -> list[Any]:
    """Build the complete processor chain for structlog.

    Args:
        config: Telemetry configuration
        log_stream: Output stream for logging

    Returns:
        List of processors for structlog

    """
    core_processors = _build_core_processors_list(config)
    formatter_processors = _build_formatter_processors_list(config.logging, log_stream)
    return cast("list[Any]", core_processors + formatter_processors)

configure_structlog_output

configure_structlog_output(
    config: TelemetryConfig, log_stream: TextIO
) -> None

Configure structlog with the complete output chain.

Parameters:

Name	Type	Description	Default
config	TelemetryConfig	Telemetry configuration	required
log_stream	TextIO	Output stream for logging	required

Source code in provide/foundation/logger/setup/processors.py

def configure_structlog_output(
    config: TelemetryConfig,
    log_stream: TextIO,
) -> None:
    """Configure structlog with the complete output chain.

    Args:
        config: Telemetry configuration
        log_stream: Output stream for logging

    """
    from provide.foundation.logger.constants import LEVEL_TO_NUMERIC

    processors = build_complete_processor_chain(config, log_stream)
    effective_level = LEVEL_TO_NUMERIC.get(config.logging.default_level, 20)

    # FilteringBoundLogger must use the minimum of all configured levels so
    # module-level overrides (e.g. auth=DEBUG when default=INFO) can reach
    # the _LevelFilter processor which evaluates per-module thresholds.
    if config.logging.module_levels:
        for module_level_str in config.logging.module_levels.values():
            module_numeric = LEVEL_TO_NUMERIC.get(module_level_str, 20)
            if module_numeric < effective_level:
                effective_level = module_numeric

    apply_structlog_configuration(processors, log_stream, effective_level)

handle_globally_disabled_setup

handle_globally_disabled_setup() -> None

Configure structlog for globally disabled telemetry (no-op mode).

Uses a null logger factory that drops all output. The processor chain must still strip Foundation-specific context to avoid errors.

Source code in provide/foundation/logger/setup/processors.py

def handle_globally_disabled_setup() -> None:
    """Configure structlog for globally disabled telemetry (no-op mode).

    Uses a null logger factory that drops all output. The processor chain
    must still strip Foundation-specific context to avoid errors.
    """

    class NullLogger:
        """Logger that silently drops all output."""

        def msg(self, message: str) -> None:
            """Drop the message."""

        def __getattr__(self, name: str) -> Any:
            """Return self for any attribute access (debug, info, etc.)."""
            return self.msg

    class NullLoggerFactory:
        """Factory that returns NullLogger instances."""

        def __call__(self, *args: Any, **kwargs: Any) -> NullLogger:
            return NullLogger()

    def strip_foundation_context(
        _logger: Any,
        _method_name: str,
        event_dict: dict[str, object],
    ) -> dict[str, object]:
        """Strip Foundation-specific bound context before rendering."""
        event_dict.pop("logger_name", None)
        event_dict.pop("_foundation_level_hint", None)
        return event_dict

    structlog.configure(
        processors=[
            strip_foundation_context,
            structlog.dev.ConsoleRenderer(),
        ],
        wrapper_class=structlog.BoundLogger,
        logger_factory=NullLoggerFactory(),
        cache_logger_on_first_use=True,
    )