aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/sentry_sdk/integrations/fastapi.py
diff options
context:
space:
mode:
Diffstat (limited to '.venv/lib/python3.12/site-packages/sentry_sdk/integrations/fastapi.py')
-rw-r--r--.venv/lib/python3.12/site-packages/sentry_sdk/integrations/fastapi.py147
1 files changed, 147 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/sentry_sdk/integrations/fastapi.py b/.venv/lib/python3.12/site-packages/sentry_sdk/integrations/fastapi.py
new file mode 100644
index 00000000..76c6adee
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/sentry_sdk/integrations/fastapi.py
@@ -0,0 +1,147 @@
+import asyncio
+from copy import deepcopy
+from functools import wraps
+
+import sentry_sdk
+from sentry_sdk.integrations import DidNotEnable
+from sentry_sdk.scope import should_send_default_pii
+from sentry_sdk.tracing import SOURCE_FOR_STYLE, TransactionSource
+from sentry_sdk.utils import (
+ transaction_from_function,
+ logger,
+)
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from typing import Any, Callable, Dict
+ from sentry_sdk._types import Event
+
+try:
+ from sentry_sdk.integrations.starlette import (
+ StarletteIntegration,
+ StarletteRequestExtractor,
+ )
+except DidNotEnable:
+ raise DidNotEnable("Starlette is not installed")
+
+try:
+ import fastapi # type: ignore
+except ImportError:
+ raise DidNotEnable("FastAPI is not installed")
+
+
+_DEFAULT_TRANSACTION_NAME = "generic FastAPI request"
+
+
+class FastApiIntegration(StarletteIntegration):
+ identifier = "fastapi"
+
+ @staticmethod
+ def setup_once():
+ # type: () -> None
+ patch_get_request_handler()
+
+
+def _set_transaction_name_and_source(scope, transaction_style, request):
+ # type: (sentry_sdk.Scope, str, Any) -> None
+ name = ""
+
+ if transaction_style == "endpoint":
+ endpoint = request.scope.get("endpoint")
+ if endpoint:
+ name = transaction_from_function(endpoint) or ""
+
+ elif transaction_style == "url":
+ route = request.scope.get("route")
+ if route:
+ path = getattr(route, "path", None)
+ if path is not None:
+ name = path
+
+ if not name:
+ name = _DEFAULT_TRANSACTION_NAME
+ source = TransactionSource.ROUTE
+ else:
+ source = SOURCE_FOR_STYLE[transaction_style]
+
+ scope.set_transaction_name(name, source=source)
+ logger.debug(
+ "[FastAPI] Set transaction name and source on scope: %s / %s", name, source
+ )
+
+
+def patch_get_request_handler():
+ # type: () -> None
+ old_get_request_handler = fastapi.routing.get_request_handler
+
+ def _sentry_get_request_handler(*args, **kwargs):
+ # type: (*Any, **Any) -> Any
+ dependant = kwargs.get("dependant")
+ if (
+ dependant
+ and dependant.call is not None
+ and not asyncio.iscoroutinefunction(dependant.call)
+ ):
+ old_call = dependant.call
+
+ @wraps(old_call)
+ def _sentry_call(*args, **kwargs):
+ # type: (*Any, **Any) -> Any
+ current_scope = sentry_sdk.get_current_scope()
+ if current_scope.transaction is not None:
+ current_scope.transaction.update_active_thread()
+
+ sentry_scope = sentry_sdk.get_isolation_scope()
+ if sentry_scope.profile is not None:
+ sentry_scope.profile.update_active_thread_id()
+
+ return old_call(*args, **kwargs)
+
+ dependant.call = _sentry_call
+
+ old_app = old_get_request_handler(*args, **kwargs)
+
+ async def _sentry_app(*args, **kwargs):
+ # type: (*Any, **Any) -> Any
+ integration = sentry_sdk.get_client().get_integration(FastApiIntegration)
+ if integration is None:
+ return await old_app(*args, **kwargs)
+
+ request = args[0]
+
+ _set_transaction_name_and_source(
+ sentry_sdk.get_current_scope(), integration.transaction_style, request
+ )
+ sentry_scope = sentry_sdk.get_isolation_scope()
+ extractor = StarletteRequestExtractor(request)
+ info = await extractor.extract_request_info()
+
+ def _make_request_event_processor(req, integration):
+ # type: (Any, Any) -> Callable[[Event, Dict[str, Any]], Event]
+ def event_processor(event, hint):
+ # type: (Event, Dict[str, Any]) -> Event
+
+ # Extract information from request
+ request_info = event.get("request", {})
+ if info:
+ if "cookies" in info and should_send_default_pii():
+ request_info["cookies"] = info["cookies"]
+ if "data" in info:
+ request_info["data"] = info["data"]
+ event["request"] = deepcopy(request_info)
+
+ return event
+
+ return event_processor
+
+ sentry_scope._name = FastApiIntegration.identifier
+ sentry_scope.add_event_processor(
+ _make_request_event_processor(request, integration)
+ )
+
+ return await old_app(*args, **kwargs)
+
+ return _sentry_app
+
+ fastapi.routing.get_request_handler = _sentry_get_request_handler