sentry_application.py

import functools
from urllib import parse

import sentry_sdk


class SentryMiddleware:
    """
    This is a middleware for django channels, sentry already provide an integration
    for django.

    this was inspired by: https://github.com/encode/sentry-asgi/blob/master
    /sentry_asgi/middleware.py
    """

    def __init__(self, app, scope):
        self.app = app(scope)
        self.scope = scope

    async def __call__(self, receive, send):
        hub = sentry_sdk.Hub.current
        with sentry_sdk.Hub(hub) as hub:
            with hub.configure_scope() as sentry_scope:
                sentry_scope.clear()
                sentry_scope.clear_breadcrumbs()
                sentry_scope.add_event_processor(self.event_processor)
                try:
                    await self.app(receive, send)
                except Exception as exc:
                    hub.capture_exception(exc)
                    raise exc from None

    def event_processor(self, event, hint):
        if self.scope["type"] in ("http", "websocket"):
            event["request"] = {
                "url": self.get_url(self.scope),
                "method": self.scope["method"],
                "headers": self.get_headers(self.scope),
                "query_string": self.get_query(self.scope),
            }
            if self.scope.get("client"):
                event["request"]["env"] = {"REMOTE_ADDR": self.scope["client"][0]}

        event["transaction"] = self.scope["path"]

        return event

    def get_url(self, scope):
        """
        Extract URL from the ASGI scope, without also including the querystring.
        """
        scheme = scope.get("scheme", "http")
        server = scope.get("server", None)
        path = scope.get("root_path", "") + scope["path"]

        for key, value in scope["headers"]:
            if key == b"host":
                host_header = value.decode("latin-1")
                return parse.urlunparse(
                    (
                        scheme,
                        host_header,
                        path,
                        None,  # path params
                        None,  # query
                        None,  # fragment
                    )
                )

        if server is not None:
            host, port = server
            default_port = {"http": 80, "https": 443, "ws": 80, "wss": 443}[scheme]
            if port != default_port:
                return parse.urlunparse(
                    (
                        scheme,
                        f"{host}:{port}",
                        path,
                        None,  # path params
                        None,  # query
                        None,  # fragment
                    )
                )
            return parse.urlunparse(
                (
                    scheme,
                    host,
                    path,
                    None,  # path params
                    None,  # query
                    None,  # fragment
                )
            )

    def get_query(self, scope):
        """
        Extract querystring from the ASGI scope, in the format that the Sentry
        protocol expects.
        """
        return parse.unquote(scope["query_string"].decode("latin-1"))

    def get_headers(self, scope):
        """
        Extract headers from the ASGI scope, in the format that the Sentry protocol
        expects.
        """
        headers = {}
        for raw_key, raw_value in scope["headers"]:
            key = raw_key.decode("latin-1")
            value = raw_value.decode("latin-1")
            if key in headers:
                headers[key] = headers[key] + ", " + value
            else:
                headers[key] = value
        return headers


def SentryApplicationMiddleware(app):
    return functools.partial(SentryMiddleware, app)