sentry_logbook_handler.py

# Tweak the existing Logging integration to work with Logbook.
# Mostly copy-paste from https://github.com/getsentry/sentry-python/blob/master/sentry_sdk/integrations/logging.py
#
# Usage:
#
#     StreamHandler(sys.stdout, level="INFO").push_application()
#     SentryErrorHandler(bubble=True, level='ERROR').push_application()
#     SentryBreadcrumbHandler(bubble=True, level='INFO').push_application()
#

import logbook
from sentry_sdk import Hub
from sentry_sdk.utils import capture_internal_exceptions, event_from_exception, current_stacktrace, to_string


class SentryErrorHandler(logbook.Handler):
    def emit(self, record):
        with capture_internal_exceptions():
            self.format(record)
            return self._emit(record)

    def _emit(self, record):
        hub = Hub.current
        if hub.client is None:
            return

        client_options = hub.client.options

        # exc_info might be None or (None, None, None)
        if record.exc_info is not None and record.exc_info[0] is not None:
            event, hint = event_from_exception(
                record.exc_info,
                client_options=client_options,
                mechanism={"type": "logbook", "handled": True},
            )
        elif record.exc_info and record.exc_info[0] is None:
            event = {}
            hint = {}
            with capture_internal_exceptions():
                event["threads"] = {
                    "values": [
                        {
                            "stacktrace": current_stacktrace(
                                client_options["with_locals"]
                            ),
                            "crashed": False,
                            "current": True,
                        }
                    ]
                }
        else:
            event = {}
            hint = {}

        hint["log_record"] = record

        event["level"] = logbook_to_event_level(record.level)
        event["logger"] = record.channel
        event["logentry"] = {"message": to_string(record.msg), "params": record.args}
        event["extra"] = _extra_from_record(record)

        hub.capture_event(event, hint=hint)


class SentryBreadcrumbHandler(logbook.Handler):
    """
    A logging handler that records breadcrumbs for each log record.
    Note that you do not have to use this class if the logging integration is enabled, which it is by default.
    """

    def emit(self, record):
        with capture_internal_exceptions():
            self.format(record)
            return self._emit(record)

    def _emit(self, record):
        Hub.current.add_breadcrumb(
            _breadcrumb_from_record(record), hint={"log_record": record}
        )


def _breadcrumb_from_record(record):
    return {
        "ty": "log",
        "level": logbook_to_event_level(record.level),
        "category": record.channel,
        "message": record.message,
        "timestamp": record.time,
        "data": _extra_from_record(record)
    }


def logbook_to_event_level(levelname):
    # type: (str) -> str
    # return {"critical": "fatal"}.get(levelname.lower(), levelname.lower())
    return logbook.get_level_name(levelname).lower()


def _extra_from_record(record):
    return {
        "lineno": record.lineno,
        "filename": record.filename,
        "function": record.func_name,
        "process": record.process,
        "process_name": record.process_name,
    }