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: boto3.py

from __future__ import absolute_import

from sentry_sdk import Hub
from sentry_sdk.consts import OP, SPANDATA
from sentry_sdk.integrations import Integration, DidNotEnable
from sentry_sdk.tracing import Span

from sentry_sdk._functools import partial
from sentry_sdk._types import TYPE_CHECKING
from sentry_sdk.utils import capture_internal_exceptions, parse_url, parse_version

if TYPE_CHECKING:
    from typing import Any
    from typing import Dict
    from typing import Optional
    from typing import Type

try:
    from botocore import __version__ as BOTOCORE_VERSION  # type: ignore
    from botocore.client import BaseClient  # type: ignore
    from botocore.response import StreamingBody  # type: ignore
    from botocore.awsrequest import AWSRequest  # type: ignore
except ImportError:
    raise DidNotEnable("botocore is not installed")

class Boto3Integration(Integration):
    identifier = "boto3"

@staticmethod
    def setup_once():
        # type: () -> None

version = parse_version(BOTOCORE_VERSION)

if version is None:
            raise DidNotEnable(
                "Unparsable botocore version: {}".format(BOTOCORE_VERSION)
            )

if version < (1, 12):
            raise DidNotEnable("Botocore 1.12 or newer is required.")

orig_init = BaseClient.__init__

def sentry_patched_init(self, *args, **kwargs):
            # type: (Type[BaseClient], *Any, **Any) -> None
            orig_init(self, *args, **kwargs)
            meta = self.meta
            service_id = meta.service_model.service_id.hyphenize()
            meta.events.register(
                "request-created",
                partial(_sentry_request_created, service_id=service_id),
            )
            meta.events.register("after-call", _sentry_after_call)
            meta.events.register("after-call-error", _sentry_after_call_error)

BaseClient.__init__ = sentry_patched_init

def _sentry_request_created(service_id, request, operation_name, **kwargs):
    # type: (str, AWSRequest, str, **Any) -> None
    hub = Hub.current
    if hub.get_integration(Boto3Integration) is None:
        return

description = "aws.%s.%s" % (service_id, operation_name)
    span = hub.start_span(
        hub=hub,
        op=OP.HTTP_CLIENT,
        description=description,
    )

with capture_internal_exceptions():
        parsed_url = parse_url(request.url, sanitize=False)
        span.set_data("aws.request.url", parsed_url.url)
        span.set_data(SPANDATA.HTTP_QUERY, parsed_url.query)
        span.set_data(SPANDATA.HTTP_FRAGMENT, parsed_url.fragment)

span.set_tag("aws.service_id", service_id)
    span.set_tag("aws.operation_name", operation_name)
    span.set_data(SPANDATA.HTTP_METHOD, request.method)

# We do it in order for subsequent http calls/retries be
    # attached to this span.
    span.__enter__()

# request.context is an open-ended data-structure
    # where we can add anything useful in request life cycle.
    request.context["_sentrysdk_span"] = span

def _sentry_after_call(context, parsed, **kwargs):
    # type: (Dict[str, Any], Dict[str, Any], **Any) -> None
    span = context.pop("_sentrysdk_span", None)  # type: Optional[Span]

# Span could be absent if the integration is disabled.
    if span is None:
        return
    span.__exit__(None, None, None)

body = parsed.get("Body")
    if not isinstance(body, StreamingBody):
        return

streaming_span = span.start_child(
        op=OP.HTTP_CLIENT_STREAM,
        description=span.description,
    )

orig_read = body.read
    orig_close = body.close

def sentry_streaming_body_read(*args, **kwargs):
        # type: (*Any, **Any) -> bytes
        try:
            ret = orig_read(*args, **kwargs)
            if not ret:
                streaming_span.finish()
            return ret
        except Exception:
            streaming_span.finish()
            raise

body.read = sentry_streaming_body_read

def sentry_streaming_body_close(*args, **kwargs):
        # type: (*Any, **Any) -> None
        streaming_span.finish()
        orig_close(*args, **kwargs)

body.close = sentry_streaming_body_close

def _sentry_after_call_error(context, exception, **kwargs):
    # type: (Dict[str, Any], Type[BaseException], **Any) -> None
    span = context.pop("_sentrysdk_span", None)  # type: Optional[Span]

# Span could be absent if the integration is disabled.
    if span is None:
        return
    span.__exit__(type(exception), exception, None)

X7ROOT File Manager

Editing: boto3.py

Upload File

Create Folder