𝗖𝗨𝗥𝗥𝗘𝗡𝗧 𝗙𝗜𝗟𝗘 : /opt/hc_python/lib/python3.12/site-packages/sentry_sdk/integrations//asgi.py
"""
An ASGI middleware.
Based on Tom Christie's `sentry-asgi <https://github.com/encode/sentry-asgi>`.
"""
import inspect
import sys
from copy import deepcopy
from functools import partial
from typing import TYPE_CHECKING
import sentry_sdk
from sentry_sdk.api import continue_trace
from sentry_sdk.consts import OP, SPANDATA
from sentry_sdk.integrations._asgi_common import (
_get_headers,
_get_ip,
_get_request_attributes,
_get_request_data,
_get_url,
)
from sentry_sdk.integrations._wsgi_common import (
DEFAULT_HTTP_METHODS_TO_CAPTURE,
nullcontext,
)
from sentry_sdk.scope import Scope, should_send_default_pii
from sentry_sdk.sessions import track_session
from sentry_sdk.traces import (
SOURCE_FOR_STYLE as SEGMENT_SOURCE_FOR_STYLE,
)
from sentry_sdk.traces import (
SegmentSource,
StreamedSpan,
)
from sentry_sdk.tracing import (
SOURCE_FOR_STYLE,
Transaction,
TransactionSource,
)
from sentry_sdk.tracing_utils import has_span_streaming_enabled
from sentry_sdk.utils import (
CONTEXTVARS_ERROR_MESSAGE,
HAS_REAL_CONTEXTVARS,
ContextVar,
_get_installed_modules,
capture_internal_exceptions,
event_from_exception,
logger,
qualname_from_function,
reraise,
transaction_from_function,
)
if TYPE_CHECKING:
from typing import Any, ContextManager, Dict, Optional, Tuple, Union
from sentry_sdk._types import Attributes, Event, Hint
from sentry_sdk.tracing import Span
_asgi_middleware_applied = ContextVar("sentry_asgi_middleware_applied")
_DEFAULT_TRANSACTION_NAME = "generic ASGI request"
TRANSACTION_STYLE_VALUES = ("endpoint", "url")
# Vendored: https://github.com/Kludex/uvicorn/blob/b224045f5900b7f766743bcb16ba9fc3adea2606/uvicorn/_compat.py#L10-L13
if sys.version_info >= (3, 14):
from inspect import iscoroutinefunction
else:
from asyncio import iscoroutinefunction
def _capture_exception(exc: "Any", mechanism_type: str = "asgi") -> None:
event, hint = event_from_exception(
exc,
client_options=sentry_sdk.get_client().options,
mechanism={"type": mechanism_type, "handled": False},
)
sentry_sdk.capture_event(event, hint=hint)
def _looks_like_asgi3(app: "Any") -> bool:
"""
Try to figure out if an application object supports ASGI3.
This is how uvicorn figures out the application version as well.
"""
if inspect.isclass(app):
return hasattr(app, "__await__")
elif inspect.isfunction(app):
return iscoroutinefunction(app)
else:
call = getattr(app, "__call__", None) # noqa
return iscoroutinefunction(call)
class SentryAsgiMiddleware:
__slots__ = (
"app",
"__call__",
"transaction_style",
"mechanism_type",
"span_origin",
"http_methods_to_capture",
)
def __init__(
self,
app: "Any",
unsafe_context_data: bool = False,
transaction_style: str = "endpoint",
mechanism_type: str = "asgi",
span_origin: str = "manual",
http_methods_to_capture: "Tuple[str, ...]" = DEFAULT_HTTP_METHODS_TO_CAPTURE,
asgi_version: "Optional[int]" = None,
) -> None:
"""
Instrument an ASGI application with Sentry. Provides HTTP/websocket
data to sent events and basic handling for exceptions bubbling up
through the middleware.
:param unsafe_context_data: Disable errors when a proper contextvars installation could not be found. We do not recommend changing this from the default.
"""
if not unsafe_context_data and not HAS_REAL_CONTEXTVARS:
# We better have contextvars or we're going to leak state between
# requests.
raise RuntimeError(
"The ASGI middleware for Sentry requires Python 3.7+ "
"or the aiocontextvars package." + CONTEXTVARS_ERROR_MESSAGE
)
if transaction_style not in TRANSACTION_STYLE_VALUES:
raise ValueError(
"Invalid value for transaction_style: %s (must be in %s)"
% (transaction_style, TRANSACTION_STYLE_VALUES)
)
asgi_middleware_while_using_starlette_or_fastapi = (
mechanism_type == "asgi" and "starlette" in _get_installed_modules()
)
if asgi_middleware_while_using_starlette_or_fastapi:
logger.warning(
"The Sentry Python SDK can now automatically support ASGI frameworks like Starlette and FastAPI. "
"Please remove 'SentryAsgiMiddleware' from your project. "
"See https://docs.sentry.io/platforms/python/guides/asgi/ for more information."
)
self.transaction_style = transaction_style
self.mechanism_type = mechanism_type
self.span_origin = span_origin
self.app = app
self.http_methods_to_capture = http_methods_to_capture
if asgi_version is None:
if _looks_like_asgi3(app):
asgi_version = 3
else:
asgi_version = 2
if asgi_version == 3:
self.__call__ = self._run_asgi3
elif asgi_version == 2:
self.__call__ = self._run_asgi2 # type: ignore
def _capture_lifespan_exception(self, exc: Exception) -> None:
"""Capture exceptions raise in application lifespan handlers.
The separate function is needed to support overriding in derived integrations that use different catching mechanisms.
"""
return _capture_exception(exc=exc, mechanism_type=self.mechanism_type)
def _capture_request_exception(self, exc: Exception) -> None:
"""Capture exceptions raised in incoming request handlers.
The separate function is needed to support overriding in derived integrations that use different catching mechanisms.
"""
return _capture_exception(exc=exc, mechanism_type=self.mechanism_type)
def _run_asgi2(self, scope: "Any") -> "Any":
async def inner(receive: "Any", send: "Any") -> "Any":
return await self._run_app(scope, receive, send, asgi_version=2)
return inner
async def _run_asgi3(self, scope: "Any", receive: "Any", send: "Any") -> "Any":
return await self._run_app(scope, receive, send, asgi_version=3)
async def _run_app(
self, scope: "Any", receive: "Any", send: "Any", asgi_version: int
) -> "Any":
is_recursive_asgi_middleware = _asgi_middleware_applied.get(False)
is_lifespan = scope["type"] == "lifespan"
if is_recursive_asgi_middleware or is_lifespan:
try:
if asgi_version == 2:
return await self.app(scope)(receive, send)
else:
return await self.app(scope, receive, send)
except Exception as exc:
suppress_chained_exceptions = (
sentry_sdk.get_client()
.options.get("_experiments", {})
.get("suppress_asgi_chained_exceptions", True)
)
if suppress_chained_exceptions:
self._capture_lifespan_exception(exc)
raise exc from None
exc_info = sys.exc_info()
with capture_internal_exceptions():
self._capture_lifespan_exception(exc)
reraise(*exc_info)
client = sentry_sdk.get_client()
span_streaming = has_span_streaming_enabled(client.options)
_asgi_middleware_applied.set(True)
try:
with sentry_sdk.isolation_scope() as sentry_scope:
with track_session(sentry_scope, session_mode="request"):
sentry_scope.clear_breadcrumbs()
sentry_scope._name = "asgi"
processor = partial(self.event_processor, asgi_scope=scope)
sentry_scope.add_event_processor(processor)
ty = scope["type"]
(
transaction_name,
transaction_source,
) = self._get_transaction_name_and_source(
self.transaction_style,
scope,
)
method = scope.get("method", "").upper()
span_ctx: "ContextManager[Union[Span, StreamedSpan, None]]"
if span_streaming:
segment: "Optional[StreamedSpan]" = None
attributes: "Attributes" = {
"sentry.span.source": getattr(
transaction_source, "value", transaction_source
),
"sentry.origin": self.span_origin,
"network.protocol.name": ty,
}
if scope.get("client") and should_send_default_pii():
sentry_scope.set_attribute(
SPANDATA.USER_IP_ADDRESS, _get_ip(scope)
)
if ty in ("http", "websocket"):
if (
ty == "websocket"
or method in self.http_methods_to_capture
):
sentry_sdk.traces.continue_trace(_get_headers(scope))
Scope.set_custom_sampling_context({"asgi_scope": scope})
attributes["sentry.op"] = f"{ty}.server"
segment = sentry_sdk.traces.start_span(
name=transaction_name,
attributes=attributes,
parent_span=None,
)
else:
sentry_sdk.traces.new_trace()
Scope.set_custom_sampling_context({"asgi_scope": scope})
attributes["sentry.op"] = OP.HTTP_SERVER
segment = sentry_sdk.traces.start_span(
name=transaction_name,
attributes=attributes,
parent_span=None,
)
span_ctx = segment or nullcontext()
else:
transaction = None
if ty in ("http", "websocket"):
if (
ty == "websocket"
or method in self.http_methods_to_capture
):
transaction = continue_trace(
_get_headers(scope),
op="{}.server".format(ty),
name=transaction_name,
source=transaction_source,
origin=self.span_origin,
)
else:
transaction = Transaction(
op=OP.HTTP_SERVER,
name=transaction_name,
source=transaction_source,
origin=self.span_origin,
)
if transaction:
transaction.set_tag("asgi.type", ty)
span_ctx = (
sentry_sdk.start_transaction(
transaction,
custom_sampling_context={"asgi_scope": scope},
)
if transaction is not None
else nullcontext()
)
with span_ctx as span:
if isinstance(span, StreamedSpan):
for attribute, value in _get_request_attributes(
scope
).items():
span.set_attribute(attribute, value)
try:
async def _sentry_wrapped_send(
event: "Dict[str, Any]",
) -> "Any":
if span is not None:
is_http_response = (
event.get("type") == "http.response.start"
and "status" in event
)
if is_http_response:
if isinstance(span, StreamedSpan):
span.status = (
"error"
if event["status"] >= 400
else "ok"
)
span.set_attribute(
"http.response.status_code",
event["status"],
)
else:
span.set_http_status(event["status"])
return await send(event)
if asgi_version == 2:
return await self.app(scope)(
receive, _sentry_wrapped_send
)
else:
return await self.app(
scope, receive, _sentry_wrapped_send
)
except Exception as exc:
suppress_chained_exceptions = (
sentry_sdk.get_client()
.options.get("_experiments", {})
.get("suppress_asgi_chained_exceptions", True)
)
if suppress_chained_exceptions:
self._capture_request_exception(exc)
raise exc from None
exc_info = sys.exc_info()
with capture_internal_exceptions():
self._capture_request_exception(exc)
reraise(*exc_info)
finally:
if isinstance(span, StreamedSpan):
already_set = (
span is not None
and span.name != _DEFAULT_TRANSACTION_NAME
and span.get_attributes().get("sentry.span.source")
in [
SegmentSource.COMPONENT.value,
SegmentSource.ROUTE.value,
SegmentSource.CUSTOM.value,
]
)
with capture_internal_exceptions():
if not already_set:
name, source = (
self._get_segment_name_and_source(
self.transaction_style, scope
)
)
span.name = name
span.set_attribute("sentry.span.source", source)
finally:
_asgi_middleware_applied.set(False)
def event_processor(
self, event: "Event", hint: "Hint", asgi_scope: "Any"
) -> "Optional[Event]":
request_data = event.get("request", {})
request_data.update(_get_request_data(asgi_scope))
event["request"] = deepcopy(request_data)
# Only set transaction name if not already set by Starlette or FastAPI (or other frameworks)
transaction = event.get("transaction")
transaction_source = (event.get("transaction_info") or {}).get("source")
already_set = (
transaction is not None
and transaction != _DEFAULT_TRANSACTION_NAME
and transaction_source
in [
TransactionSource.COMPONENT,
TransactionSource.ROUTE,
TransactionSource.CUSTOM,
]
)
if not already_set:
name, source = self._get_transaction_name_and_source(
self.transaction_style, asgi_scope
)
event["transaction"] = name
event["transaction_info"] = {"source": source}
return event
# Helper functions.
#
# Note: Those functions are not public API. If you want to mutate request
# data to your liking it's recommended to use the `before_send` callback
# for that.
def _get_transaction_name_and_source(
self: "SentryAsgiMiddleware", transaction_style: str, asgi_scope: "Any"
) -> "Tuple[str, str]":
name = None
source = SOURCE_FOR_STYLE[transaction_style]
ty = asgi_scope.get("type")
if transaction_style == "endpoint":
endpoint = asgi_scope.get("endpoint")
# Webframeworks like Starlette mutate the ASGI env once routing is
# done, which is sometime after the request has started. If we have
# an endpoint, overwrite our generic transaction name.
if endpoint:
name = transaction_from_function(endpoint) or ""
else:
name = _get_url(asgi_scope, "http" if ty == "http" else "ws", host=None)
source = TransactionSource.URL
elif transaction_style == "url":
# FastAPI includes the route object in the scope to let Sentry extract the
# path from it for the transaction name
route = asgi_scope.get("route")
if route:
path = getattr(route, "path", None)
if path is not None:
name = path
else:
name = _get_url(asgi_scope, "http" if ty == "http" else "ws", host=None)
source = TransactionSource.URL
if name is None:
name = _DEFAULT_TRANSACTION_NAME
source = TransactionSource.ROUTE
return name, source
return name, source
def _get_segment_name_and_source(
self: "SentryAsgiMiddleware", segment_style: str, asgi_scope: "Any"
) -> "Tuple[str, str]":
name = None
source = SEGMENT_SOURCE_FOR_STYLE[segment_style].value
ty = asgi_scope.get("type")
if segment_style == "endpoint":
endpoint = asgi_scope.get("endpoint")
# Webframeworks like Starlette mutate the ASGI env once routing is
# done, which is sometime after the request has started. If we have
# an endpoint, overwrite our generic transaction name.
if endpoint:
name = qualname_from_function(endpoint) or ""
else:
name = _get_url(asgi_scope, "http" if ty == "http" else "ws", host=None)
source = SegmentSource.URL.value
elif segment_style == "url":
# FastAPI includes the route object in the scope to let Sentry extract the
# path from it for the transaction name
route = asgi_scope.get("route")
if route:
path = getattr(route, "path", None)
if path is not None:
name = path
else:
name = _get_url(asgi_scope, "http" if ty == "http" else "ws", host=None)
source = SegmentSource.URL.value
if name is None:
name = _DEFAULT_TRANSACTION_NAME
source = SegmentSource.ROUTE.value
return name, source
return name, source
