# 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.
"""
**Deprecated** primitives to keep context global but thread (and greenlet)
local.
See `thread-local`, but please use :doc:`contextvars` instead.
.. deprecated:: 22.1.0
"""
from __future__ import annotations
import contextlib
import sys
import threading
import uuid
import warnings
from typing import Any, Generator, Iterator, TypeVar
import structlog
from ._config import BoundLoggerLazyProxy
from .typing 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 # pragma: no cover
ThreadLocal = _determine_threadlocal()
def _deprecated() -> None:
"""
Raise a warning with best-effort stacklevel adjustment.
"""
callsite = ""
with contextlib.suppress(Exception):
f = sys._getframe()
callsite = f.f_back.f_back.f_globals[ # type: ignore[union-attr]
"__name__"
]
# Avoid double warnings if TL functions call themselves.
if callsite == "structlog.threadlocal":
return
stacklevel = 3
# If a function is used as a decorator, we need to add two stack levels.
# This logic will probably break eventually, but it's not worth any more
# complexity.
if callsite == "contextlib":
stacklevel += 2
warnings.warn(
"`structlog.threadlocal` is deprecated, please use "
"`structlog.contextvars` instead.",
DeprecationWarning,
stacklevel=stacklevel,
)
[docs]
def wrap_dict(dict_class: type[Context]) -> type[Context]:
"""
Wrap a dict-like class and return the resulting class.
The wrapped class and used to keep global in the current thread.
Args:
dict_class: Class used for keeping context.
.. deprecated:: 22.1.0
"""
_deprecated()
Wrapped = type(
"WrappedDict-" + str(uuid.uuid4()), (_ThreadLocalDictWrapper,), {}
)
Wrapped._tl = ThreadLocal() # type: ignore[attr-defined]
Wrapped._dict_class = dict_class # type: ignore[attr-defined]
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.
Args:
logger (structlog.typing.BindableLogger):
A logger with *possibly* thread local state.
Returns:
:class:`~structlog.BoundLogger` with an immutable context.
.. deprecated:: 22.1.0
"""
_deprecated()
if isinstance(logger, BoundLoggerLazyProxy):
logger = logger.bind()
try:
ctx = logger._context._tl.dict_.__class__( # type: ignore[union-attr]
logger._context._dict # type: ignore[union-attr]
)
bl = logger.__class__(
logger._logger, # type: ignore[attr-defined, call-arg]
processors=logger._processors, # type: ignore[attr-defined]
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.
Only works with `structlog.threadlocal.wrap_dict`-based contexts.
Use :func:`~structlog.threadlocal.bound_threadlocal` for new code.
.. deprecated:: 22.1.0
"""
_deprecated()
if isinstance(logger, BoundLoggerLazyProxy):
logger = logger.bind()
saved = as_immutable(logger)._context
try:
yield logger.bind(**tmp_values) # type: ignore[misc]
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.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: object) -> bool:
# Same class == same dictionary
return self.__class__ == other.__class__
def __ne__(self, other: object) -> 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:
return getattr(self._dict, name)
_CONTEXT = threading.local()
[docs]
def get_threadlocal() -> Context:
"""
Return a copy of the current thread-local context.
.. versionadded:: 21.2.0
.. deprecated:: 22.1.0
"""
_deprecated()
return _get_context().copy()
[docs]
def get_merged_threadlocal(bound_logger: BindableLogger) -> Context:
"""
Return a copy of the current thread-local context merged with the context
from *bound_logger*.
.. versionadded:: 21.2.0
.. deprecated:: 22.1.0
"""
_deprecated()
ctx = _get_context().copy()
ctx.update(structlog.get_context(bound_logger))
return ctx
[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_threadlocal_context`` and that
name is still kept around for backward compatibility.
.. deprecated:: 22.1.0
"""
_deprecated()
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
.. deprecated:: 22.1.0
"""
_deprecated()
_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
.. deprecated:: 22.1.0
"""
_deprecated()
_get_context().update(kw)
[docs]
def unbind_threadlocal(*keys: str) -> None:
"""
Tries to remove bound *keys* from threadlocal logging context if present.
.. versionadded:: 20.1.0
.. deprecated:: 22.1.0
"""
_deprecated()
context = _get_context()
for key in keys:
context.pop(key, None)
[docs]
@contextlib.contextmanager
def bound_threadlocal(**kw: Any) -> Generator[None, None, None]:
"""
Bind *kw* to the current thread-local context. Unbind or restore *kw*
afterwards. Do **not** affect other keys.
Can be used as a context manager or decorator.
.. versionadded:: 21.4.0
.. deprecated:: 22.1.0
"""
_deprecated()
context = get_threadlocal()
saved = {k: context[k] for k in context.keys() & kw.keys()}
bind_threadlocal(**kw)
try:
yield
finally:
unbind_threadlocal(*kw.keys())
bind_threadlocal(**saved)
def _get_context() -> Context:
try:
return _CONTEXT.context
except AttributeError:
_CONTEXT.context = {}
return _CONTEXT.context