diff options
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.py | 147 |
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 |