Skip to content

Processors

provide.foundation.logger.setup.processors

TODO: Add module docstring.

Classes

Functions

apply_structlog_configuration 

apply_structlog_configuration(
    processors: list[Any], log_stream: TextIO
) -> None

Apply the processor configuration to structlog.

Parameters:

Name Type Description Default
processors list[Any]

List of processors to configure

 required
log_stream TextIO

Output stream for logging

 required
Source code in provide/foundation/logger/setup/processors.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def apply_structlog_configuration(processors: list[Any], log_stream: TextIO) -> None:
    """Apply the processor configuration to structlog.

    Args:
        processors: List of processors to configure
        log_stream: Output stream for logging

    """
    # 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=cast("type[structlog.types.BindableLogger]", structlog.BoundLogger),
        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 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
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 
70
71
72
73
74
75
76
77
78
79
80
81
82
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

    """
    processors = build_complete_processor_chain(config, log_stream)
    apply_structlog_configuration(processors, log_stream)

handle_globally_disabled_setup 

handle_globally_disabled_setup() -> None

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

Source code in provide/foundation/logger/setup/processors.py 
85
86
87
88
89
90
91
def handle_globally_disabled_setup() -> None:
    """Configure structlog for globally disabled telemetry (no-op mode)."""
    structlog.configure(
        processors=[],
        logger_factory=structlog.ReturnLoggerFactory(),
        cache_logger_on_first_use=True,
    )