# 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.
"""
Processors useful regardless of the logging framework.
"""
import datetime
import json
import operator
import sys
import time
from typing import (
Any,
Callable,
Dict,
List,
Optional,
Sequence,
TextIO,
Tuple,
Union,
)
from ._frames import (
_find_first_app_frame_and_name,
_format_exception,
_format_stack,
)
from ._log_levels import _NAME_TO_LEVEL, add_log_level
from .types import EventDict, ExcInfo, WrappedLogger
__all__ = [
"_NAME_TO_LEVEL", # some people rely on it being here
"KeyValueRenderer",
"TimeStamper",
"add_log_level",
"UnicodeEncoder",
"UnicodeDecoder",
"JSONRenderer",
"format_exc_info",
"ExceptionPrettyPrinter",
"StackInfoRenderer",
]
[docs]class KeyValueRenderer:
"""
Render ``event_dict`` as a list of ``Key=repr(Value)`` pairs.
:param sort_keys: Whether to sort keys when formatting.
:param key_order: List of keys that should be rendered in this exact
order. Missing keys will be rendered as ``None``, extra keys depending
on *sort_keys* and the dict class.
:param drop_missing: When ``True``, extra keys in *key_order* will be
dropped rather than rendered as ``None``.
:param repr_native_str: When ``True``, :func:`repr()` is also applied
to native strings (i.e. unicode on Python 3 and bytes on Python 2).
Setting this to ``False`` is useful if you want to have human-readable
non-ASCII output on Python 2.
.. versionadded:: 0.2.0 *key_order*
.. versionadded:: 16.1.0 *drop_missing*
.. versionadded:: 17.1.0 *repr_native_str*
"""
def __init__(
self,
sort_keys: bool = False,
key_order: Optional[Sequence[str]] = None,
drop_missing: bool = False,
repr_native_str: bool = True,
):
# Use an optimized version for each case.
if key_order and sort_keys is True:
def ordered_items(event_dict: EventDict) -> List[Tuple[str, Any]]:
items = []
for key in key_order: # type: ignore
value = event_dict.pop(key, None)
if value is not None or not drop_missing:
items.append((key, value))
items += sorted(event_dict.items())
return items
elif key_order:
def ordered_items(event_dict: EventDict) -> List[Tuple[str, Any]]:
items = []
for key in key_order: # type: ignore
value = event_dict.pop(key, None)
if value is not None or not drop_missing:
items.append((key, value))
items += event_dict.items()
return items
elif sort_keys:
def ordered_items(event_dict: EventDict) -> List[Tuple[str, Any]]:
return sorted(event_dict.items())
else:
ordered_items = operator.methodcaller("items")
self._ordered_items = ordered_items
if repr_native_str is True:
self._repr = repr
else:
def _repr(inst: Any) -> str:
if isinstance(inst, str):
return inst
else:
return repr(inst)
self._repr = _repr
def __call__(
self, _: WrappedLogger, __: str, event_dict: EventDict
) -> str:
return " ".join(
k + "=" + self._repr(v) for k, v in self._ordered_items(event_dict)
)
[docs]class UnicodeEncoder:
"""
Encode unicode values in ``event_dict``.
:param encoding: Encoding to encode to (default: ``"utf-8"``).
:param errors: How to cope with encoding errors (default
``"backslashreplace"``).
Useful if you're running Python 2 as otherwise ``u"abc"`` will be rendered
as ``'u"abc"'``.
Just put it in the processor chain before the renderer.
"""
_encoding: str
_errors: str
def __init__(
self, encoding: str = "utf-8", errors: str = "backslashreplace"
) -> None:
self._encoding = encoding
self._errors = errors
def __call__(
self, logger: WrappedLogger, name: str, event_dict: EventDict
) -> EventDict:
for key, value in event_dict.items():
if isinstance(value, str):
event_dict[key] = value.encode(self._encoding, self._errors)
return event_dict
[docs]class UnicodeDecoder:
"""
Decode byte string values in ``event_dict``.
:param encoding: Encoding to decode from (default: ``"utf-8"``).
:param errors: How to cope with encoding errors (default:
``"replace"``).
Useful if you're running Python 3 as otherwise ``b"abc"`` will be rendered
as ``'b"abc"'``.
Just put it in the processor chain before the renderer.
.. versionadded:: 15.4.0
"""
_encoding: str
_errors: str
def __init__(
self, encoding: str = "utf-8", errors: str = "replace"
) -> None:
self._encoding = encoding
self._errors = errors
def __call__(
self, logger: WrappedLogger, name: str, event_dict: EventDict
) -> EventDict:
for key, value in event_dict.items():
if isinstance(value, bytes):
event_dict[key] = value.decode(self._encoding, self._errors)
return event_dict
[docs]class JSONRenderer:
"""
Render the ``event_dict`` using ``serializer(event_dict, **json_kw)``.
:param json_kw: Are passed unmodified to *serializer*. If *default*
is passed, it will disable support for ``__structlog__``-based
serialization.
:param serializer: A :func:`json.dumps`-compatible callable that
will be used to format the string. This can be used to use alternative
JSON encoders like `simplejson
<https://pypi.org/project/simplejson/>`_ or `RapidJSON
<https://pypi.org/project/python-rapidjson/>`_ (faster but Python
3-only) (default: :func:`json.dumps`).
.. versionadded:: 0.2.0
Support for ``__structlog__`` serialization method.
.. versionadded:: 15.4.0
*serializer* parameter.
.. versionadded:: 18.2.0
Serializer's *default* parameter can be overwritten now.
"""
def __init__(
self,
serializer: Callable[..., Union[str, bytes]] = json.dumps,
**dumps_kw: Any
) -> None:
dumps_kw.setdefault("default", _json_fallback_handler)
self._dumps_kw = dumps_kw
self._dumps = serializer
def __call__(
self, logger: WrappedLogger, name: str, event_dict: EventDict
) -> Union[str, bytes]:
"""
The return type of this depends on the return type of self._dumps.
"""
return self._dumps(event_dict, **self._dumps_kw)
def _json_fallback_handler(obj: Any) -> Any:
"""
Serialize custom datatypes and pass the rest to __structlog__ & repr().
"""
# circular imports :(
from structlog.threadlocal import _ThreadLocalDictWrapper
if isinstance(obj, _ThreadLocalDictWrapper):
return obj._dict
else:
try:
return obj.__structlog__()
except AttributeError:
return repr(obj)
[docs]class TimeStamper:
"""
Add a timestamp to ``event_dict``.
:param fmt: strftime format string, or ``"iso"`` for `ISO 8601
<https://en.wikipedia.org/wiki/ISO_8601>`_, or `None` for a `UNIX
timestamp <https://en.wikipedia.org/wiki/Unix_time>`_.
:param utc: Whether timestamp should be in UTC or local time.
:param key: Target key in *event_dict* for added timestamps.
.. versionchanged:: 19.2 Can be pickled now.
"""
__slots__ = ("_stamper", "fmt", "utc", "key")
def __init__(
self,
fmt: Optional[str] = None,
utc: bool = True,
key: str = "timestamp",
) -> None:
self.fmt, self.utc, self.key = fmt, utc, key
self._stamper = _make_stamper(fmt, utc, key)
def __call__(
self, logger: WrappedLogger, name: str, event_dict: EventDict
) -> EventDict:
return self._stamper(event_dict)
def __getstate__(self) -> Dict[str, Any]:
return {"fmt": self.fmt, "utc": self.utc, "key": self.key}
def __setstate__(self, state: Dict[str, Any]) -> None:
self.fmt = state["fmt"]
self.utc = state["utc"]
self.key = state["key"]
self._stamper = _make_stamper(**state)
def _make_stamper(
fmt: Optional[str], utc: bool, key: str
) -> Callable[[EventDict], EventDict]:
"""
Create a stamper function.
"""
if fmt is None and not utc:
raise ValueError("UNIX timestamps are always UTC.")
now = getattr(datetime.datetime, "utcnow" if utc else "now")
if fmt is None:
def stamper_unix(event_dict: EventDict) -> EventDict:
event_dict[key] = time.time()
return event_dict
return stamper_unix
elif fmt.upper() == "ISO":
def stamper_iso_local(event_dict: EventDict) -> EventDict:
event_dict[key] = now().isoformat()
return event_dict
def stamper_iso_utc(event_dict: EventDict) -> EventDict:
event_dict[key] = now().isoformat() + "Z"
return event_dict
if utc:
return stamper_iso_utc
else:
return stamper_iso_local
def stamper_fmt(event_dict: EventDict) -> EventDict:
event_dict[key] = now().strftime(fmt)
return event_dict
return stamper_fmt
def _figure_out_exc_info(v: Any) -> ExcInfo:
"""
Depending on the Python version will try to do the smartest thing possible
to transform *v* into an ``exc_info`` tuple.
"""
if isinstance(v, BaseException):
return (v.__class__, v, v.__traceback__)
elif isinstance(v, tuple):
return v # type: ignore
elif v:
return sys.exc_info() # type: ignore
return v
[docs]class ExceptionPrettyPrinter:
"""
Pretty print exceptions and remove them from the ``event_dict``.
:param file: Target file for output (default: ``sys.stdout``).
This processor is mostly for development and testing so you can read
exceptions properly formatted.
It behaves like format_exc_info` except it removes the exception
data from the event dictionary after printing it.
It's tolerant to having `format_exc_info` in front of itself in the
processor chain but doesn't require it. In other words, it handles both
``exception`` as well as ``exc_info`` keys.
.. versionadded:: 0.4.0
.. versionchanged:: 16.0.0
Added support for passing exceptions as ``exc_info`` on Python 3.
"""
def __init__(self, file: Optional[TextIO] = None) -> None:
if file is not None:
self._file = file
else:
self._file = sys.stdout
def __call__(
self, logger: WrappedLogger, name: str, event_dict: EventDict
) -> EventDict:
exc = event_dict.pop("exception", None)
if exc is None:
exc_info = _figure_out_exc_info(event_dict.pop("exc_info", None))
if exc_info:
exc = _format_exception(exc_info)
if exc:
print(exc, file=self._file)
return event_dict
[docs]class StackInfoRenderer:
"""
Add stack information with key ``stack`` if ``stack_info`` is `True`.
Useful when you want to attach a stack dump to a log entry without
involving an exception.
It works analogously to the *stack_info* argument of the Python 3 standard
library logging but works on both 2 and 3.
.. versionadded:: 0.4.0
"""
def __call__(
self, logger: WrappedLogger, name: str, event_dict: EventDict
) -> EventDict:
if event_dict.pop("stack_info", None):
event_dict["stack"] = _format_stack(
_find_first_app_frame_and_name()[0]
)
return event_dict