Source code for eventtracking.backends.logger

"""Event tracker backend that saves events to a python logger."""


from datetime import datetime
from datetime import date
import logging
import json

from pytz import UTC

MAX_EVENT_SIZE = 1024  # 1 KB


[docs]class LoggerBackend: """ Event tracker backend that uses a python logger. Events are logged to the INFO level as JSON strings. """ def __init__(self, **kwargs): """ Event tracker backend that uses a python logger. `name` is an identifier for the logger, which should have been configured using the default python mechanisms. """ name = kwargs.get('name', None) self.max_event_size = kwargs.get('max_event_size', MAX_EVENT_SIZE) self.event_logger = logging.getLogger(name) level = kwargs.get('level', 'info') self.log = getattr(self.event_logger, level.lower())
[docs] def send(self, event): """Send the event to the standard python logger""" event_str = json.dumps(event, cls=DateTimeJSONEncoder) # TODO: do something smarter than simply dropping the event on # the floor. if self.max_event_size is None or len(event_str) <= self.max_event_size: self.log(event_str)
[docs]class DateTimeJSONEncoder(json.JSONEncoder): """JSON encoder aware of datetime.datetime and datetime.date objects"""
[docs] def default(self, obj): # pylint: disable=arguments-renamed """ Serialize datetime and date objects of iso format. datatime objects are converted to UTC. """ if isinstance(obj, datetime): if obj.tzinfo is None: # Localize to UTC naive datetime objects obj = UTC.localize(obj) # pylint: disable=no-value-for-parameter else: # Convert to UTC datetime objects from other timezones obj = obj.astimezone(UTC) return obj.isoformat() elif isinstance(obj, date): return obj.isoformat() return super().default(obj)