Skip to content

infrastructure.logging.console_adapter

src.infrastructure.logging.console_adapter

Console logging adapter (development/testing).

Outputs structured logs to stdout using structlog. - Development: human-readable console renderer with colors - Testing/CI: JSON renderer for machine parsing

Implementation intentionally does NOT inherit from LoggerProtocol (PEP 544 structural subtyping). Any object with the same call signatures is compatible with LoggerProtocol.

Classes

ConsoleAdapter

Console logger for development and testing environments.

Parameters:

Name Type Description Default
use_json bool

JSON output when True (CI/testing), human-readable when False (dev).

 False
Source code in src/infrastructure/logging/console_adapter.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 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
126
127
128
129
130
131
132
class ConsoleAdapter:
    """Console logger for development and testing environments.

    Args:
        use_json (bool): JSON output when True (CI/testing), human-readable when False (dev).
    """

    def __init__(self, *, use_json: bool = False) -> None:
        """Initialize the console adapter.

        Args:
            use_json (bool): JSON output when True (CI/testing), human-readable when False (dev).
        """
        processors: list[structlog.types.Processor] = [
            structlog.stdlib.add_log_level,
            structlog.processors.TimeStamper(fmt="iso", utc=True),
            structlog.processors.StackInfoRenderer(),
        ]

        if use_json:
            processors.append(structlog.processors.JSONRenderer())
        else:
            processors.append(structlog.dev.ConsoleRenderer(colors=True))

        structlog.configure(
            processors=processors,
            wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
            context_class=dict,
            logger_factory=structlog.PrintLoggerFactory(file=sys.stdout),
            cache_logger_on_first_use=True,
        )

        self._logger = structlog.get_logger()

    def debug(self, message: str, /, **context: Any) -> None:
        """Log a debug message.

        Args:
            message (str): Message text.
            **context: Structured key-value context.
        """
        self._logger.debug(message, **context)

    def info(self, message: str, /, **context: Any) -> None:
        """Log an info message.

        Args:
            message (str): Message text.
            **context: Structured key-value context.
        """
        self._logger.info(message, **context)

    def warning(self, message: str, /, **context: Any) -> None:
        """Log a warning message.

        Args:
            message (str): Message text.
            **context: Structured key-value context.
        """
        self._logger.warning(message, **context)

    def error(
        self, message: str, /, *, error: Exception | None = None, **context: Any
    ) -> None:
        """Log an error message with optional exception details.

        Args:
            message (str): Message text.
            error (Exception | None): Optional exception instance.
            **context: Structured key-value context.
        """
        if error is not None:
            context["error_type"] = type(error).__name__
            context["error_message"] = str(error)
        self._logger.error(message, **context)

    def critical(
        self, message: str, /, *, error: Exception | None = None, **context: Any
    ) -> None:
        """Log a critical message with optional exception details.

        Args:
            message (str): Message text.
            error (Exception | None): Optional exception instance.
            **context: Structured key-value context.
        """
        if error is not None:
            context["error_type"] = type(error).__name__
            context["error_message"] = str(error)
        self._logger.critical(message, **context)

    def bind(self, **context: Any) -> ConsoleAdapter:
        """Return new adapter with bound context.

        Args:
            **context: Context to bind to all subsequent logs.

        Returns:
            ConsoleAdapter: New adapter instance with bound context.
        """
        bound_adapter = ConsoleAdapter.__new__(ConsoleAdapter)
        bound_adapter._logger = self._logger.bind(**context)
        return bound_adapter

    def with_context(self, **context: Any) -> ConsoleAdapter:
        """Return new adapter with bound context (alias for bind).

        Args:
            **context: Context to bind to all subsequent logs.

        Returns:
            ConsoleAdapter: New adapter instance with bound context.
        """
        return self.bind(**context)
Functions
__init__ 
__init__(*, use_json: bool = False) -> None

Parameters:

Name Type Description Default
use_json bool

JSON output when True (CI/testing), human-readable when False (dev).

 False
Source code in src/infrastructure/logging/console_adapter.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(self, *, use_json: bool = False) -> None:
    """Initialize the console adapter.

    Args:
        use_json (bool): JSON output when True (CI/testing), human-readable when False (dev).
    """
    processors: list[structlog.types.Processor] = [
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.StackInfoRenderer(),
    ]

    if use_json:
        processors.append(structlog.processors.JSONRenderer())
    else:
        processors.append(structlog.dev.ConsoleRenderer(colors=True))

    structlog.configure(
        processors=processors,
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(file=sys.stdout),
        cache_logger_on_first_use=True,
    )

    self._logger = structlog.get_logger()
debug 
debug(message: str, /, **context: Any) -> None

Log a debug message.

Parameters:

Name Type Description Default
message str

Message text.

 required
**context Any

Structured key-value context.

 {}
Source code in src/infrastructure/logging/console_adapter.py 
53
54
55
56
57
58
59
60
def debug(self, message: str, /, **context: Any) -> None:
    """Log a debug message.

    Args:
        message (str): Message text.
        **context: Structured key-value context.
    """
    self._logger.debug(message, **context)
info 
info(message: str, /, **context: Any) -> None

Log an info message.

Parameters:

Name Type Description Default
message str

Message text.

 required
**context Any

Structured key-value context.

 {}
Source code in src/infrastructure/logging/console_adapter.py 
62
63
64
65
66
67
68
69
def info(self, message: str, /, **context: Any) -> None:
    """Log an info message.

    Args:
        message (str): Message text.
        **context: Structured key-value context.
    """
    self._logger.info(message, **context)
warning 
warning(message: str, /, **context: Any) -> None

Log a warning message.

Parameters:

Name Type Description Default
message str

Message text.

 required
**context Any

Structured key-value context.

 {}
Source code in src/infrastructure/logging/console_adapter.py 
71
72
73
74
75
76
77
78
def warning(self, message: str, /, **context: Any) -> None:
    """Log a warning message.

    Args:
        message (str): Message text.
        **context: Structured key-value context.
    """
    self._logger.warning(message, **context)
error 
error(
    message: str,
    /,
    *,
    error: Exception | None = None,
    **context: Any,
) -> None

Log an error message with optional exception details.

Parameters:

Name Type Description Default
message str

Message text.

 required
error Exception | None

Optional exception instance.

 None
**context Any

Structured key-value context.

 {}
Source code in src/infrastructure/logging/console_adapter.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def error(
    self, message: str, /, *, error: Exception | None = None, **context: Any
) -> None:
    """Log an error message with optional exception details.

    Args:
        message (str): Message text.
        error (Exception | None): Optional exception instance.
        **context: Structured key-value context.
    """
    if error is not None:
        context["error_type"] = type(error).__name__
        context["error_message"] = str(error)
    self._logger.error(message, **context)
critical 
critical(
    message: str,
    /,
    *,
    error: Exception | None = None,
    **context: Any,
) -> None

Log a critical message with optional exception details.

Parameters:

Name Type Description Default
message str

Message text.

 required
error Exception | None

Optional exception instance.

 None
**context Any

Structured key-value context.

 {}
Source code in src/infrastructure/logging/console_adapter.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def critical(
    self, message: str, /, *, error: Exception | None = None, **context: Any
) -> None:
    """Log a critical message with optional exception details.

    Args:
        message (str): Message text.
        error (Exception | None): Optional exception instance.
        **context: Structured key-value context.
    """
    if error is not None:
        context["error_type"] = type(error).__name__
        context["error_message"] = str(error)
    self._logger.critical(message, **context)
bind 
bind(**context: Any) -> ConsoleAdapter

Return new adapter with bound context.

Parameters:

Name Type Description Default
**context Any

Context to bind to all subsequent logs.

 {}

Returns:

Name Type Description
ConsoleAdapter ConsoleAdapter

New adapter instance with bound context.
Source code in src/infrastructure/logging/console_adapter.py 
110
111
112
113
114
115
116
117
118
119
120
121
def bind(self, **context: Any) -> ConsoleAdapter:
    """Return new adapter with bound context.

    Args:
        **context: Context to bind to all subsequent logs.

    Returns:
        ConsoleAdapter: New adapter instance with bound context.
    """
    bound_adapter = ConsoleAdapter.__new__(ConsoleAdapter)
    bound_adapter._logger = self._logger.bind(**context)
    return bound_adapter
with_context 
with_context(**context: Any) -> ConsoleAdapter

Return new adapter with bound context (alias for bind).

Parameters:

Name Type Description Default
**context Any

Context to bind to all subsequent logs.

 {}

Returns:

Name Type Description
ConsoleAdapter ConsoleAdapter

New adapter instance with bound context.
Source code in src/infrastructure/logging/console_adapter.py 
123
124
125
126
127
128
129
130
131
132
def with_context(self, **context: Any) -> ConsoleAdapter:
    """Return new adapter with bound context (alias for bind).

    Args:
        **context: Context to bind to all subsequent logs.

    Returns:
        ConsoleAdapter: New adapter instance with bound context.
    """
    return self.bind(**context)