Source code for structlog.threadlocal

# 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.

"""
Primitives to keep context global but thread (and greenlet) local.

See `thread-local`.
"""
import contextlib
import threading
import uuid

from typing import Any, Dict, Generator, Iterator, Type, TypeVar

from ._config import BoundLoggerLazyProxy
from .types import BindableLogger, Context, EventDict, WrappedLogger


def _determine_threadlocal() -> Type[Any]:
    """
    Return a dict-like threadlocal storage depending on whether we run with
    greenlets or not.
    """
    try:
        from ._greenlets import GreenThreadLocal
    except ImportError:
        from threading import local

        return local

    return GreenThreadLocal


ThreadLocal = _determine_threadlocal()


[docs]def wrap_dict(dict_class: Type[EventDict]) -> Type[EventDict]: """ Wrap a dict-like class and return the resulting class. The wrapped class and used to keep global in the current thread. :param dict_class: Class used for keeping context. """ Wrapped = type( "WrappedDict-" + str(uuid.uuid4()), (_ThreadLocalDictWrapper,), {} ) Wrapped._tl = ThreadLocal() # type: ignore Wrapped._dict_class = dict_class # type: ignore return Wrapped
TLLogger = TypeVar("TLLogger", bound=BindableLogger)
[docs]def as_immutable(logger: TLLogger) -> TLLogger: """ Extract the context from a thread local logger into an immutable logger. :param structlog.types.BindableLogger logger: A logger with *possibly* thread local state. :returns: :class:`~structlog.BoundLogger` with an immutable context. """ if isinstance(logger, BoundLoggerLazyProxy): logger = logger.bind() # type: ignore try: ctx = logger._context._tl.dict_.__class__( # type: ignore logger._context._dict # type: ignore ) bl = logger.__class__( logger._logger, # type: ignore processors=logger._processors, # type: ignore context={}, ) bl._context = ctx return bl except AttributeError: return logger
[docs]@contextlib.contextmanager def tmp_bind( logger: TLLogger, **tmp_values: Any ) -> Generator[TLLogger, None, None]: """ Bind *tmp_values* to *logger* & memorize current state. Rewind afterwards. """ saved = as_immutable(logger)._context try: yield logger.bind(**tmp_values) # type: ignore finally: logger._context.clear() logger._context.update(saved)
class _ThreadLocalDictWrapper: """ Wrap a dict-like class and keep the state *global* but *thread-local*. Attempts to re-initialize only updates the wrapped dictionary. Useful for short-lived threaded applications like requests in web app. Use :func:`wrap` to instantiate and use :func:`structlog._loggers.BoundLogger.new` to clear the context. """ _tl: Any _dict_class: Type[Dict[str, Any]] def __init__(self, *args: Any, **kw: Any) -> None: """ We cheat. A context dict gets never recreated. """ if args and isinstance(args[0], self.__class__): # our state is global, no need to look at args[0] if it's of our # class self._dict.update(**kw) else: self._dict.update(*args, **kw) @property def _dict(self) -> Context: """ Return or create and return the current context. """ try: return self.__class__._tl.dict_ except AttributeError: self.__class__._tl.dict_ = self.__class__._dict_class() return self.__class__._tl.dict_ def __repr__(self) -> str: return f"<{self.__class__.__name__}({self._dict!r})>" def __eq__(self, other: Any) -> bool: # Same class == same dictionary return self.__class__ == other.__class__ def __ne__(self, other: Any) -> bool: return not self.__eq__(other) # Proxy methods necessary for structlog. # Dunder methods don't trigger __getattr__ so we need to proxy by hand. def __iter__(self) -> Iterator[str]: return self._dict.__iter__() def __setitem__(self, key: str, value: Any) -> None: self._dict[key] = value def __delitem__(self, key: str) -> None: self._dict.__delitem__(key) def __len__(self) -> int: return self._dict.__len__() def __getattr__(self, name: str) -> Any: method = getattr(self._dict, name) return method _CONTEXT = threading.local()
[docs]def merge_threadlocal( logger: WrappedLogger, method_name: str, event_dict: EventDict ) -> EventDict: """ A processor that merges in a global (thread-local) context. Use this as your first processor in :func:`structlog.configure` to ensure thread-local context is included in all log calls. .. versionadded:: 19.2.0 .. versionchanged:: 20.1.0 This function used to be called ``merge_threalocal_context`` and that name is still kept around for backward compatibility. """ context = _get_context().copy() context.update(event_dict) return context
# Alias that shouldn't be used anymore. merge_threadlocal_context = merge_threadlocal
[docs]def clear_threadlocal() -> None: """ Clear the thread-local context. The typical use-case for this function is to invoke it early in request-handling code. .. versionadded:: 19.2.0 """ _CONTEXT.context = {}
[docs]def bind_threadlocal(**kw: Any) -> None: """ Put keys and values into the thread-local context. Use this instead of :func:`~structlog.BoundLogger.bind` when you want some context to be global (thread-local). .. versionadded:: 19.2.0 """ _get_context().update(kw)
def unbind_threadlocal(*keys: str) -> None: """ Tries to remove bound *keys* from threadlocal logging context if present. .. versionadded:: 20.1.0 """ context = _get_context() for key in keys: context.pop(key, None) def _get_context() -> Context: try: return _CONTEXT.context except AttributeError: _CONTEXT.context = {} return _CONTEXT.context