Source code for structlog._base

# SPDX-License-Identifier: MIT OR Apache-2.0
# This file is dual licensed under the terms of the Apache License, Version
# 2.0, and the MIT License.  See the LICENSE file in the root of this
# repository for complete details.

"""
Logger wrapper and helper class.
"""

from typing import Any, Dict, Iterable, Mapping, Optional, Sequence, Tuple

from structlog.exceptions import DropEvent

from .types import BindableLogger, Context, Processor, WrappedLogger


[docs]class BoundLoggerBase: """ Immutable context carrier. Doesn't do any actual logging; examples for useful subclasses are: - the generic `BoundLogger` that can wrap anything, - `structlog.stdlib.BoundLogger`. - `structlog.twisted.BoundLogger`, See also `custom-wrappers`. """ _logger: WrappedLogger """ Wrapped logger. .. note:: Despite underscore available **read-only** to custom wrapper classes. See also `custom-wrappers`. """ def __init__( self, logger: WrappedLogger, processors: Iterable[Processor], context: Context, ): self._logger = logger self._processors = processors self._context = context def __repr__(self) -> str: return "<{}(context={!r}, processors={!r})>".format( self.__class__.__name__, self._context, self._processors ) def __eq__(self, other: Any) -> bool: try: if self._context == other._context: return True else: return False except AttributeError: return False def __ne__(self, other: Any) -> bool: return not self.__eq__(other)
[docs] def bind(self, **new_values: Any) -> "BoundLoggerBase": """ Return a new logger with *new_values* added to the existing ones. """ return self.__class__( self._logger, self._processors, self._context.__class__(self._context, **new_values), )
[docs] def unbind(self, *keys: str) -> "BoundLoggerBase": """ Return a new logger with *keys* removed from the context. :raises KeyError: If the key is not part of the context. """ bl = self.bind() for key in keys: del bl._context[key] return bl
def try_unbind(self, *keys: str) -> "BoundLoggerBase": """ Like :meth:`unbind`, but best effort: missing keys are ignored. .. versionadded:: 18.2.0 """ bl = self.bind() for key in keys: bl._context.pop(key, None) return bl
[docs] def new(self, **new_values: Any) -> "BoundLoggerBase": """ Clear context and binds *initial_values* using `bind`. Only necessary with dict implementations that keep global state like those wrapped by `structlog.threadlocal.wrap_dict` when threads are re-used. """ self._context.clear() return self.bind(**new_values)
# Helper methods for sub-classing concrete BoundLoggers.
[docs] def _process_event( self, method_name: str, event: Optional[str], event_kw: Dict[str, Any] ) -> Tuple[Sequence[Any], Mapping[str, Any]]: """ Combines creates an ``event_dict`` and runs the chain. Call it to combine your *event* and *context* into an event_dict and process using the processor chain. :param method_name: The name of the logger method. Is passed into the processors. :param event: The event -- usually the first positional argument to a logger. :param event_kw: Additional event keywords. For example if someone calls ``log.msg("foo", bar=42)``, *event* would to be ``"foo"`` and *event_kw* ``{"bar": 42}``. :raises: `structlog.DropEvent` if log entry should be dropped. :raises: `ValueError` if the final processor doesn't return a str, bytes, bytearray, tuple, or a dict. :returns: `tuple` of ``(*args, **kw)`` .. note:: Despite underscore available to custom wrapper classes. See also `custom-wrappers`. .. versionchanged:: 14.0.0 Allow final processor to return a `dict`. .. versionchanged:: 20.2.0 Allow final processor to return `bytes`. .. versionchanged:: 21.2.0 Allow final processor to return a `bytearray`. """ # We're typing it as Any, because processors can return more than an # EventDict. event_dict: Any = self._context.copy() event_dict.update(**event_kw) if event is not None: event_dict["event"] = event for proc in self._processors: event_dict = proc(self._logger, method_name, event_dict) if isinstance(event_dict, (str, bytes, bytearray)): return (event_dict,), {} elif isinstance(event_dict, tuple): # In this case we assume that the last processor returned a tuple # of ``(args, kwargs)`` and pass it right through. return event_dict # type: ignore elif isinstance(event_dict, dict): return (), event_dict else: raise ValueError( "Last processor didn't return an appropriate value. Valid " "return values are a dict, a tuple of (args, kwargs), bytes, " "or a str." )
[docs] def _proxy_to_logger( self, method_name: str, event: Optional[str] = None, **event_kw: Any ) -> Any: """ Run processor chain on event & call *method_name* on wrapped logger. DRY convenience method that runs :func:`_process_event`, takes care of handling :exc:`structlog.DropEvent`, and finally calls *method_name* on :attr:`_logger` with the result. :param method_name: The name of the method that's going to get called. Technically it should be identical to the method the user called because it also get passed into processors. :param event: The event -- usually the first positional argument to a logger. :param event_kw: Additional event keywords. For example if someone calls ``log.msg("foo", bar=42)``, *event* would to be ``"foo"`` and *event_kw* ``{"bar": 42}``. .. note:: Despite underscore available to custom wrapper classes. See also `custom-wrappers`. """ try: args, kw = self._process_event(method_name, event, event_kw) return getattr(self._logger, method_name)(*args, **kw) except DropEvent: return
[docs]def get_context(bound_logger: BindableLogger) -> Context: """ Return *bound_logger*'s context. The type of *bound_logger* and the type returned depend on your configuration. :param bound_logger: The bound logger whose context you want. :returns: The *actual* context from *bound_logger*. It is *not* copied first. .. versionadded:: 20.2 """ # This probably will get more complicated in the future. return bound_logger._context