# 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.
"""
Global state department. Don't reload this module or everything breaks.
"""
from __future__ import annotations
import os
import sys
import warnings
from typing import Any, Callable, Iterable, Sequence, Type, cast
from ._native import make_filtering_bound_logger
from ._output import PrintLoggerFactory
from .contextvars import merge_contextvars
from .dev import ConsoleRenderer, _has_colors, set_exc_info
from .processors import StackInfoRenderer, TimeStamper, add_log_level
from .typing import BindableLogger, Context, Processor, WrappedLogger
"""
Any changes to these defaults must be reflected in:
- `getting-started`.
- structlog.stdlib.recreate_defaults()'s docstring.
"""
_BUILTIN_DEFAULT_PROCESSORS: Sequence[Processor] = [
merge_contextvars,
add_log_level,
StackInfoRenderer(),
set_exc_info,
TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
ConsoleRenderer(
colors=os.environ.get("NO_COLOR", "") == ""
and (
os.environ.get("FORCE_COLOR", "") != ""
or (
_has_colors
and sys.stdout is not None
and hasattr(sys.stdout, "isatty")
and sys.stdout.isatty()
)
)
),
]
_BUILTIN_DEFAULT_CONTEXT_CLASS = cast(Type[Context], dict)
_BUILTIN_DEFAULT_WRAPPER_CLASS = make_filtering_bound_logger(0)
_BUILTIN_DEFAULT_LOGGER_FACTORY = PrintLoggerFactory()
_BUILTIN_CACHE_LOGGER_ON_FIRST_USE = False
class _Configuration:
"""
Global defaults.
"""
is_configured: bool = False
default_processors: Iterable[Processor] = _BUILTIN_DEFAULT_PROCESSORS[:]
default_context_class: type[Context] = _BUILTIN_DEFAULT_CONTEXT_CLASS
default_wrapper_class: Any = _BUILTIN_DEFAULT_WRAPPER_CLASS
logger_factory: Callable[..., WrappedLogger] = (
_BUILTIN_DEFAULT_LOGGER_FACTORY
)
cache_logger_on_first_use: bool = _BUILTIN_CACHE_LOGGER_ON_FIRST_USE
_CONFIG = _Configuration()
"""
Global defaults used when arguments to `wrap_logger` are omitted.
"""
[docs]
def get_config() -> dict[str, Any]:
"""
Get a dictionary with the current configuration.
.. note::
Changes to the returned dictionary do *not* affect *structlog*.
.. versionadded: 18.1.0
"""
return {
"processors": _CONFIG.default_processors,
"context_class": _CONFIG.default_context_class,
"wrapper_class": _CONFIG.default_wrapper_class,
"logger_factory": _CONFIG.logger_factory,
"cache_logger_on_first_use": _CONFIG.cache_logger_on_first_use,
}
[docs]
def get_logger(*args: Any, **initial_values: Any) -> Any:
"""
Convenience function that returns a logger according to configuration.
>>> from structlog import get_logger
>>> log = get_logger(y=23)
>>> log.info("hello", x=42)
y=23 x=42 event='hello'
Args:
args:
*Optional* positional arguments that are passed unmodified to the
logger factory. Therefore it depends on the factory what they
mean.
initial_values: Values that are used to pre-populate your contexts.
Returns:
A proxy that creates a correctly configured bound logger when
necessary. The type of that bound logger depends on your configuration
and is `structlog.BoundLogger` by default.
See `configuration` for details.
If you prefer CamelCase, there's an alias for your reading pleasure:
`structlog.getLogger`.
.. versionadded:: 0.4.0 *args*
"""
return wrap_logger(None, logger_factory_args=args, **initial_values)
getLogger = get_logger # noqa: N816
"""
CamelCase alias for `structlog.get_logger`.
This function is supposed to be in every source file -- we don't want it to
stick out like a sore thumb in frameworks like Twisted or Zope.
"""
[docs]
def wrap_logger(
logger: WrappedLogger | None,
processors: Iterable[Processor] | None = None,
wrapper_class: type[BindableLogger] | None = None,
context_class: type[Context] | None = None,
cache_logger_on_first_use: bool | None = None,
logger_factory_args: Iterable[Any] | None = None,
**initial_values: Any,
) -> Any:
"""
Create a new bound logger for an arbitrary *logger*.
Default values for *processors*, *wrapper_class*, and *context_class* can
be set using `configure`.
If you set an attribute here, `configure` calls have *no* effect for the
*respective* attribute.
In other words: selective overwriting of the defaults while keeping some
*is* possible.
Args:
initial_values: Values that are used to pre-populate your contexts.
logger_factory_args:
Values that are passed unmodified as ``*logger_factory_args`` to
the logger factory if not `None`.
Returns:
A proxy that creates a correctly configured bound logger when
necessary.
See `configure` for the meaning of the rest of the arguments.
.. versionadded:: 0.4.0 *logger_factory_args*
"""
return BoundLoggerLazyProxy(
logger,
wrapper_class=wrapper_class,
processors=processors,
context_class=context_class,
cache_logger_on_first_use=cache_logger_on_first_use,
initial_values=initial_values,
logger_factory_args=logger_factory_args,
)
[docs]
def reset_defaults() -> None:
"""
Resets global default values to builtin defaults.
`is_configured` starts returning `False` afterwards.
"""
_CONFIG.is_configured = False
_CONFIG.default_processors = _BUILTIN_DEFAULT_PROCESSORS[:]
_CONFIG.default_wrapper_class = _BUILTIN_DEFAULT_WRAPPER_CLASS
_CONFIG.default_context_class = _BUILTIN_DEFAULT_CONTEXT_CLASS
_CONFIG.logger_factory = _BUILTIN_DEFAULT_LOGGER_FACTORY
_CONFIG.cache_logger_on_first_use = _BUILTIN_CACHE_LOGGER_ON_FIRST_USE
class BoundLoggerLazyProxy:
"""
Instantiates a bound logger on first usage.
Takes both configuration and instantiation parameters into account.
The only points where a bound logger changes state are ``bind()``,
``unbind()``, and ``new()`` and that return the actual ``BoundLogger``.
If and only if configuration says so, that actual bound logger is cached on
first usage.
.. versionchanged:: 0.4.0 Added support for *logger_factory_args*.
"""
# fulfill BindableLogger protocol without carrying accidental state
@property
def _context(self) -> dict[str, str]:
return self._initial_values
def __init__(
self,
logger: WrappedLogger | None,
wrapper_class: type[BindableLogger] | None = None,
processors: Iterable[Processor] | None = None,
context_class: type[Context] | None = None,
cache_logger_on_first_use: bool | None = None,
initial_values: dict[str, Any] | None = None,
logger_factory_args: Any = None,
) -> None:
self._logger = logger
self._wrapper_class = wrapper_class
self._processors = processors
self._context_class = context_class
self._cache_logger_on_first_use = cache_logger_on_first_use
self._initial_values = initial_values or {}
self._logger_factory_args = logger_factory_args or ()
def __repr__(self) -> str:
return (
f"<BoundLoggerLazyProxy(logger={self._logger!r}, wrapper_class="
f"{self._wrapper_class!r}, processors={self._processors!r}, "
f"context_class={self._context_class!r}, "
f"initial_values={self._initial_values!r}, "
f"logger_factory_args={self._logger_factory_args!r})>"
)
def bind(self, **new_values: Any) -> BindableLogger:
"""
Assemble a new BoundLogger from arguments and configuration.
"""
if self._context_class:
ctx = self._context_class(self._initial_values)
else:
ctx = _CONFIG.default_context_class(self._initial_values)
_logger = self._logger
if not _logger:
_logger = _CONFIG.logger_factory(*self._logger_factory_args)
if self._processors is None:
procs = _CONFIG.default_processors
else:
procs = self._processors
cls = self._wrapper_class or _CONFIG.default_wrapper_class
# Looks like Protocols ignore definitions of __init__ so we have to
# silence Mypy here.
logger = cls(
_logger,
processors=procs,
context=ctx, # type: ignore[call-arg]
)
def finalized_bind(**new_values: Any) -> BindableLogger:
"""
Use cached assembled logger to bind potentially new values.
"""
if new_values:
return logger.bind(**new_values)
return logger
if self._cache_logger_on_first_use is True or (
self._cache_logger_on_first_use is None
and _CONFIG.cache_logger_on_first_use is True
):
self.bind = finalized_bind # type: ignore[method-assign]
return finalized_bind(**new_values)
def unbind(self, *keys: str) -> BindableLogger:
"""
Same as bind, except unbind *keys* first.
In our case that could be only initial values.
"""
return self.bind().unbind(*keys)
def try_unbind(self, *keys: str) -> BindableLogger:
return self.bind().try_unbind(*keys)
def new(self, **new_values: Any) -> BindableLogger:
"""
Clear context, then bind.
"""
if self._context_class:
self._context_class().clear()
else:
_CONFIG.default_context_class().clear()
return self.bind(**new_values)
def __getattr__(self, name: str) -> Any:
"""
If a logging method if called on a lazy proxy, we have to create an
ephemeral BoundLogger first.
"""
if name == "__isabstractmethod__":
raise AttributeError
bl = self.bind()
return getattr(bl, name)
def __getstate__(self) -> dict[str, Any]:
"""
Our __getattr__ magic makes this necessary.
"""
return self.__dict__
def __setstate__(self, state: dict[str, Any]) -> None:
"""
Our __getattr__ magic makes this necessary.
"""
for k, v in state.items():
setattr(self, k, v)