Source code for structlog.processors

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

from __future__ import absolute_import, division, print_function

import calendar
import datetime
import json
import operator
import sys
import time

from structlog._compat import unicode_type
from structlog._frames import (
    _find_first_app_frame_and_name,
    _format_exception,
    _format_stack,
)


[docs]class KeyValueRenderer(object): """ Render `event_dict` as a list of ``Key=repr(Value)`` pairs. :param bool sort_keys: Whether to sort keys when formatting. :param list 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. >>> from structlog.processors import KeyValueRenderer >>> KeyValueRenderer(sort_keys=True)(None, None, {'a': 42, 'b': [1, 2, 3]}) 'a=42 b=[1, 2, 3]' >>> KeyValueRenderer(key_order=['b', 'a'])(None, None, ... {'a': 42, 'b': [1, 2, 3]}) 'b=[1, 2, 3] a=42' .. versionadded:: 0.2.0 `key_order` """ def __init__(self, sort_keys=False, key_order=None): # Use an optimized version for each case. if key_order and sort_keys: def ordered_items(event_dict): items = [] for key in key_order: value = event_dict.pop(key, None) items.append((key, value)) items += sorted(event_dict.items()) return items elif key_order: def ordered_items(event_dict): items = [] for key in key_order: value = event_dict.pop(key, None) items.append((key, value)) items += event_dict.items() return items elif sort_keys: def ordered_items(event_dict): return sorted(event_dict.items()) else: ordered_items = operator.methodcaller('items') self._ordered_items = ordered_items def __call__(self, _, __, event_dict): return ' '.join(k + '=' + repr(v) for k, v in self._ordered_items(event_dict))
[docs]class UnicodeEncoder(object): """ Encode unicode values in `event_dict`. :param str encoding: Encoding to encode to (default: ``'utf-8'``. :param str errors: How to cope with encoding errors (default ``'backslashreplace'``). Useful for :class:`KeyValueRenderer` if you don't want to see u-prefixes: >>> from structlog.processors import KeyValueRenderer, UnicodeEncoder >>> KeyValueRenderer()(None, None, {'foo': u'bar'}) "foo=u'bar'" >>> KeyValueRenderer()(None, None, ... UnicodeEncoder()(None, None, {'foo': u'bar'})) "foo='bar'" or :class:`JSONRenderer` and :class:`structlog.twisted.JSONRenderer` to make sure user-supplied strings don't break the renderer. Just put it in the processor chain before the renderer. """ def __init__(self, encoding='utf-8', errors='backslashreplace'): self._encoding = encoding self._errors = errors def __call__(self, logger, name, event_dict): for key, value in event_dict.items(): if isinstance(value, unicode_type): event_dict[key] = value.encode(self._encoding, self._errors) return event_dict
[docs]class JSONRenderer(object): """ Render the `event_dict` using `json.dumps(event_dict, **json_kw)`. :param json_kw: Are passed unmodified to `json.dumps()`. >>> from structlog.processors import JSONRenderer >>> JSONRenderer(sort_keys=True)(None, None, {'a': 42, 'b': [1, 2, 3]}) '{"a": 42, "b": [1, 2, 3]}' Bound objects are attempted to be serialize using a ``__structlog__`` method. If none is defined, ``repr()`` is used: >>> class C1(object): ... def __structlog__(self): ... return ['C1!'] ... def __repr__(self): ... return '__structlog__ took precedence' >>> class C2(object): ... def __repr__(self): ... return 'No __structlog__, so this is used.' >>> from structlog.processors import JSONRenderer >>> JSONRenderer(sort_keys=True)(None, None, {'c1': C1(), 'c2': C2()}) '{"c1": ["C1!"], "c2": "No __structlog__, so this is used."}' Please note that additionally to strings, you can also return any type the standard library JSON module knows about -- like in this example a list. .. versionchanged:: 0.2.0 Added support for ``__structlog__`` serialization method. """ def __init__(self, **dumps_kw): self._dumps_kw = dumps_kw def __call__(self, logger, name, event_dict): return json.dumps(event_dict, cls=_JSONFallbackEncoder, **self._dumps_kw)
class _JSONFallbackEncoder(json.JSONEncoder): """ Serialize custom datatypes and pass the rest to __structlog__ & repr(). """ def default(self, obj): """ Serialize obj with repr(obj) as fallback. """ # 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]def format_exc_info(logger, name, event_dict): """ Replace an `exc_info` field by an `exception` string field: If *event_dict* contains the key ``exc_info``, there are two possible behaviors: - If the value is a tuple, render it into the key ``exception``. - If the value true but no tuple, obtain exc_info ourselves and render that. If there is no ``exc_info`` key, the *event_dict* is not touched. This behavior is analogue to the one of the stdlib's logging. >>> from structlog.processors import format_exc_info >>> try: ... raise ValueError ... except ValueError: ... format_exc_info(None, None, {'exc_info': True})# doctest: +ELLIPSIS {'exception': 'Traceback (most recent call last):... """ exc_info = event_dict.pop('exc_info', None) if exc_info: if not isinstance(exc_info, tuple): exc_info = sys.exc_info() event_dict['exception'] = _format_exception(exc_info) return event_dict
[docs]class TimeStamper(object): """ Add a timestamp to `event_dict`. .. note:: You probably want to let OS tools take care of timestamping. See also :doc:`logging-best-practices`. :param str format: 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 bool utc: Whether timestamp should be in UTC or local time. :param str key: Target key in `event_dict` for added timestamps. >>> from structlog.processors import TimeStamper >>> TimeStamper()(None, None, {}) # doctest: +SKIP {'timestamp': 1378994017} >>> TimeStamper(fmt='iso')(None, None, {}) # doctest: +SKIP {'timestamp': '2013-09-12T13:54:26.996778Z'} >>> TimeStamper(fmt='%Y', key='year')(None, None, {}) # doctest: +SKIP {'year': '2013'} """ def __new__(cls, fmt=None, utc=True, key='timestamp'): if fmt is None and not utc: raise ValueError('UNIX timestamps are always UTC.') now_method = getattr(datetime.datetime, 'utcnow' if utc else 'now') if fmt is None: def stamper(self, _, __, event_dict): event_dict[key] = calendar.timegm(time.gmtime()) return event_dict elif fmt.upper() == 'ISO': if utc: def stamper(self, _, __, event_dict): event_dict[key] = now_method().isoformat() + 'Z' return event_dict else: def stamper(self, _, __, event_dict): event_dict[key] = now_method().isoformat() return event_dict else: def stamper(self, _, __, event_dict): event_dict[key] = now_method().strftime(fmt) return event_dict return type('TimeStamper', (object,), {'__call__': stamper})()
[docs]class ExceptionPrettyPrinter(object): """ Pretty print exceptions and remove them from the `event_dict`. :param file 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 :func:`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 """ def __init__(self, file=None): if file is not None: self._file = file else: self._file = sys.stdout def __call__(self, logger, name, event_dict): exc = event_dict.pop('exception', None) if exc is None: exc_info = event_dict.pop('exc_info', None) if exc_info: if not isinstance(exc_info, tuple): exc_info = sys.exc_info() exc = _format_exception(exc_info) if exc: print(exc, file=self._file) return event_dict
[docs]class StackInfoRenderer(object): """ 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, name, event_dict): if event_dict.pop('stack_info', None): event_dict['stack'] = _format_stack( _find_first_app_frame_and_name()[0] ) return event_dict