File Manager

X7ROOT File Manager

Current Path: /opt/cloudlinux/venv/lib/python3.11/site-packages/sentry_sdk/integrations

opt / cloudlinux / venv / lib / python3.11 / site-packages / sentry_sdk / integrations /

ðŸ“ ..
ðŸ“„ __init__.py(6.67 KB)
ðŸ“ __pycache__
ðŸ“„ _wsgi_common.py(4.43 KB)
ðŸ“„ aiohttp.py(11.28 KB)
ðŸ“„ argv.py(963 B)
ðŸ“„ arq.py(6.58 KB)
ðŸ“„ asgi.py(11.54 KB)
ðŸ“„ asyncio.py(2.98 KB)
ðŸ“„ atexit.py(1.8 KB)
ðŸ“„ aws_lambda.py(15.44 KB)
ðŸ“„ beam.py(5.56 KB)
ðŸ“„ boto3.py(4.44 KB)
ðŸ“„ bottle.py(6.32 KB)
ðŸ“„ celery.py(18.65 KB)
ðŸ“„ chalice.py(4.66 KB)
ðŸ“„ cloud_resource_context.py(6.6 KB)
ðŸ“„ dedupe.py(1.16 KB)
ðŸ“ django
ðŸ“„ excepthook.py(2.21 KB)
ðŸ“„ executing.py(1.99 KB)
ðŸ“„ falcon.py(7.8 KB)
ðŸ“„ fastapi.py(4.39 KB)
ðŸ“„ flask.py(7.72 KB)
ðŸ“„ gcp.py(8.02 KB)
ðŸ“„ gnu_backtrace.py(2.86 KB)
ðŸ“ grpc
ðŸ“„ httpx.py(4.89 KB)
ðŸ“„ huey.py(4.59 KB)
ðŸ“„ logging.py(8.97 KB)
ðŸ“„ loguru.py(2.98 KB)
ðŸ“„ modules.py(2.06 KB)
ðŸ“ opentelemetry
ðŸ“„ pure_eval.py(4.45 KB)
ðŸ“„ pymongo.py(5.87 KB)
ðŸ“„ pyramid.py(7.27 KB)
ðŸ“„ quart.py(7.2 KB)
ðŸ“ redis
ðŸ“„ rq.py(5.28 KB)
ðŸ“„ sanic.py(11.06 KB)
ðŸ“„ serverless.py(1.93 KB)
ðŸ“„ socket.py(2.88 KB)
ðŸ“ spark
ðŸ“„ sqlalchemy.py(4.14 KB)
ðŸ“„ starlette.py(22.67 KB)
ðŸ“„ starlite.py(9.85 KB)
ðŸ“„ stdlib.py(8.06 KB)
ðŸ“„ threading.py(2.87 KB)
ðŸ“„ tornado.py(7.17 KB)
ðŸ“„ trytond.py(1.7 KB)
ðŸ“„ wsgi.py(9.36 KB)

Editing: fastapi.py

import asyncio
from copy import deepcopy

from sentry_sdk._types import TYPE_CHECKING
from sentry_sdk.hub import Hub, _should_send_default_pii
from sentry_sdk.integrations import DidNotEnable
from sentry_sdk.tracing import SOURCE_FOR_STYLE, TRANSACTION_SOURCE_ROUTE
from sentry_sdk.utils import transaction_from_function

if TYPE_CHECKING:
    from typing import Any, Callable, Dict
    from sentry_sdk.scope import Scope

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: (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 = TRANSACTION_SOURCE_ROUTE
    else:
        source = SOURCE_FOR_STYLE[transaction_style]

scope.set_transaction_name(name, source=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

def _sentry_call(*args, **kwargs):
                # type: (*Any, **Any) -> Any
                hub = Hub.current
                with hub.configure_scope() as sentry_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
            hub = Hub.current
            integration = hub.get_integration(FastApiIntegration)
            if integration is None:
                return await old_app(*args, **kwargs)

with hub.configure_scope() as sentry_scope:
                request = args[0]

_set_transaction_name_and_source(
                    sentry_scope, integration.transaction_style, request
                )

extractor = StarletteRequestExtractor(request)
                info = await extractor.extract_request_info()

def _make_request_event_processor(req, integration):
                    # type: (Any, Any) -> Callable[[Dict[str, Any], Dict[str, Any]], Dict[str, Any]]
                    def event_processor(event, hint):
                        # type: (Dict[str, Any], Dict[str, Any]) -> Dict[str, Any]

# 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

X7ROOT File Manager

Editing: fastapi.py

Upload File

Create Folder