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