Skip to content

logger

logger

Structured logging with key=value and JSON output support.

Wraps the standard library logging module to provide structured output in two formats: human-readable key=value pairs (default) and machine-parseable JSON for production/cloud environments.

Values containing spaces, equals signs, or quotes are automatically escaped and wrapped in double quotes. Long values are truncated to a configurable maximum length.

The StructuredFormatter is a stdlib logging.Formatter that reads structured data from the structured_kv extra field (attached by Logger) and appends it as key=value pairs. When installed on the root handler, it unifies output from both Logger and plain logging.getLogger() calls used in the models and utils layers.

Examples:

from bigbrotr.core.logger import Logger

logger = Logger("finder")
logger.info("started", cycle=1, count=42)
# Output: started cycle=1 count=42

json_logger = Logger("finder", json_output=True)
json_logger.info("started", cycle=1)
# Output: {"message": "started", "cycle": 1}
See Also

BaseService: Creates a Logger instance for each service using SERVICE_NAME. format_kv_pairs(): Standalone formatting utility used by both Logger and StructuredFormatter.

Classes

StructuredFormatter

Bases: Formatter

Formats all log records as structured key=value output.

Reads structured data from the structured_kv extra field (attached by Logger) and appends it as key=value pairs via format_kv_pairs(). When no structured_kv is present (e.g. plain logging.getLogger() calls from models/utils), the message is emitted as-is with the same level name message prefix for consistency.

See Also

Logger: Attaches the structured_kv extra field to log records. format_kv_pairs(): Formatting utility used by the format() method.

Logger 

Logger(
    name: str,
    *,
    json_output: bool = False,
    max_value_length: int | None = None,
)

Structured logger that appends keyword arguments as extra fields.

Wraps a standard logging.Logger and formats keyword arguments as either key=value pairs or JSON, depending on configuration. All public methods mirror the standard logging API with an added **kwargs parameter.

Examples:

logger = Logger("finder")
logger.info("cycle_completed", cycle=1, duration=2.5)
# Output: cycle_completed cycle=1 duration=2.5

logger.info("relay_found", url="wss://relay.example.com")
# Output: relay_found url=wss://relay.example.com
Note

This class intentionally does not use __slots__. Adding __slots__ would break unittest.mock.patch.object() which needs to set arbitrary attributes on the instance during testing.

See Also

StructuredFormatter: Stdlib formatter that reads the structured_kv extra field attached by this class. format_kv_pairs(): Standalone formatting utility. BaseService: Creates a Logger instance per service via Logger(SERVICE_NAME).

Initialize a structured logger.

Parameters:

  • name (str) –

    Logger name, typically the service or module name. Maps to the underlying logging.getLogger(name) call.

  • json_output (bool, default: False ) –

    If True, emit JSON objects instead of key=value pairs. JSON output includes timestamp (ISO 8601), level, and service fields for compatibility with log aggregators.

  • max_value_length (int | None, default: None ) –

    Maximum character length for individual values before truncation. Defaults to 1000.

Source code in src/bigbrotr/core/logger.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def __init__(
    self,
    name: str,
    *,
    json_output: bool = False,
    max_value_length: int | None = None,
) -> None:
    """Initialize a structured logger.

    Args:
        name: Logger name, typically the service or module name.
            Maps to the underlying ``logging.getLogger(name)`` call.
        json_output: If ``True``, emit JSON objects instead of
            key=value pairs. JSON output includes ``timestamp``
            (ISO 8601), ``level``, and ``service`` fields for
            compatibility with log aggregators.
        max_value_length: Maximum character length for individual values
            before truncation. Defaults to ``1000``.
    """
    if max_value_length is None:
        max_value_length = self._DEFAULT_MAX_VALUE_LENGTH
    self._logger = logging.getLogger(name)
    self._json_output = json_output
    self._max_value_length = max_value_length
Functions
debug 
debug(msg: str, **kwargs: Any) -> None

Log a DEBUG level message with optional key=value pairs.

Source code in src/bigbrotr/core/logger.py 
209
210
211
212
213
214
def debug(self, msg: str, **kwargs: Any) -> None:
    """Log a DEBUG level message with optional key=value pairs."""
    if self._json_output:
        self._logger.debug(self._format_json(msg, "debug", kwargs))
    else:
        self._logger.debug(msg, extra=self._make_extra(kwargs))
info 
info(msg: str, **kwargs: Any) -> None

Log an INFO level message with optional key=value pairs.

Source code in src/bigbrotr/core/logger.py 
216
217
218
219
220
221
def info(self, msg: str, **kwargs: Any) -> None:
    """Log an INFO level message with optional key=value pairs."""
    if self._json_output:
        self._logger.info(self._format_json(msg, "info", kwargs))
    else:
        self._logger.info(msg, extra=self._make_extra(kwargs))
warning 
warning(msg: str, **kwargs: Any) -> None

Log a WARNING level message with optional key=value pairs.

Source code in src/bigbrotr/core/logger.py 
223
224
225
226
227
228
def warning(self, msg: str, **kwargs: Any) -> None:
    """Log a WARNING level message with optional key=value pairs."""
    if self._json_output:
        self._logger.warning(self._format_json(msg, "warning", kwargs))
    else:
        self._logger.warning(msg, extra=self._make_extra(kwargs))
error 
error(msg: str, **kwargs: Any) -> None

Log an ERROR level message with optional key=value pairs.

Source code in src/bigbrotr/core/logger.py 
230
231
232
233
234
235
def error(self, msg: str, **kwargs: Any) -> None:
    """Log an ERROR level message with optional key=value pairs."""
    if self._json_output:
        self._logger.error(self._format_json(msg, "error", kwargs))
    else:
        self._logger.error(msg, extra=self._make_extra(kwargs))
critical 
critical(msg: str, **kwargs: Any) -> None

Log a CRITICAL level message with optional key=value pairs.

Source code in src/bigbrotr/core/logger.py 
237
238
239
240
241
242
def critical(self, msg: str, **kwargs: Any) -> None:
    """Log a CRITICAL level message with optional key=value pairs."""
    if self._json_output:
        self._logger.critical(self._format_json(msg, "critical", kwargs))
    else:
        self._logger.critical(msg, extra=self._make_extra(kwargs))
exception 
exception(msg: str, **kwargs: Any) -> None

Log an ERROR level message with exception traceback and optional key=value pairs.

Source code in src/bigbrotr/core/logger.py 
244
245
246
247
248
249
def exception(self, msg: str, **kwargs: Any) -> None:
    """Log an ERROR level message with exception traceback and optional key=value pairs."""
    if self._json_output:
        self._logger.exception(self._format_json(msg, "error", kwargs))
    else:
        self._logger.exception(msg, extra=self._make_extra(kwargs))

Functions

format_kv_pairs 

format_kv_pairs(
    kwargs: dict[str, Any],
    max_value_length: int | None = 1000,
    prefix: str = " ",
) -> str

Format a dictionary as space-separated key=value pairs.

Used by Logger and StructuredFormatter to produce consistent structured output. Values are truncated to max_value_length characters, and values containing whitespace, equals signs, or quotes are automatically escaped and quoted.

Parameters:

  • kwargs (dict[str, Any]) –

    Key-value pairs to format.

  • max_value_length (int | None, default: 1000 ) –

    Maximum characters per value before truncation. Pass None to disable truncation.

  • prefix (str, default: ' ' ) –

    String prepended to the output (default: single space).

Returns:

  • str –

    Formatted string, e.g. ' key1=value1 key2="value with spaces"'.

  • str –

    Returns empty string if kwargs is empty.

See Also

Logger: Primary consumer of this function.

Source code in src/bigbrotr/core/logger.py 
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
def format_kv_pairs(
    kwargs: dict[str, Any],
    max_value_length: int | None = 1000,
    prefix: str = " ",
) -> str:
    """Format a dictionary as space-separated key=value pairs.

    Used by [Logger][bigbrotr.core.logger.Logger] and
    [StructuredFormatter][bigbrotr.core.logger.StructuredFormatter] to
    produce consistent structured output. Values are truncated to
    ``max_value_length`` characters, and values containing whitespace,
    equals signs, or quotes are automatically escaped and quoted.

    Args:
        kwargs: Key-value pairs to format.
        max_value_length: Maximum characters per value before truncation.
            Pass ``None`` to disable truncation.
        prefix: String prepended to the output (default: single space).

    Returns:
        Formatted string, e.g. ``' key1=value1 key2="value with spaces"'``.
        Returns empty string if ``kwargs`` is empty.

    See Also:
        [Logger][bigbrotr.core.logger.Logger]: Primary consumer of this
            function.
    """
    if not kwargs:
        return ""

    parts = []
    for k, v in kwargs.items():
        s = str(v)
        if max_value_length and len(s) > max_value_length:
            s = s[:max_value_length] + f"...<truncated {len(s) - max_value_length} chars>"
        # Quote values containing whitespace or characters that would break parsing
        if not s or " " in s or "=" in s or '"' in s or "'" in s:
            escaped = s.replace("\\", "\\\\").replace('"', '\\"')
            parts.append(f'{k}="{escaped}"')
        else:
            parts.append(f"{k}={s}")

    return prefix + " ".join(parts) if parts else ""