Skip to content

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
) -> 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 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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 
69
70
71
72
73
74
75
76
77
78
79
80
81
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).

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 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
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,
    )