# Source code for sentry_sdk.client

    import json
    import os
    import platform
    import random
    import socket
    import sys
    import uuid
    import warnings
    from collections.abc import Iterable, Mapping
    from datetime import datetime, timezone
    from importlib import import_module
    from typing import TYPE_CHECKING, Dict, List, cast, overload

    from sentry_sdk._compat import check_uwsgi_thread_support
    from sentry_sdk._metrics_batcher import MetricsBatcher
    from sentry_sdk._span_batcher import SpanBatcher
    from sentry_sdk.consts import (
        DEFAULT_MAX_VALUE_LENGTH,
        DEFAULT_OPTIONS,
        INSTRUMENTER,
        SPANDATA,
        SPANSTATUS,
        VERSION,
        ClientConstructor,
    )
    from sentry_sdk.envelope import Envelope, Item, PayloadRef
    from sentry_sdk.integrations import _DEFAULT_INTEGRATIONS, setup_integrations
    from sentry_sdk.integrations.dedupe import DedupeIntegration
    from sentry_sdk.monitor import Monitor
    from sentry_sdk.profiler.continuous_profiler import setup_continuous_profiler
    from sentry_sdk.profiler.transaction_profiler import (
        Profile,
        has_profiling_enabled,
        setup_profiler,
    )
    from sentry_sdk.scrubber import EventScrubber
    from sentry_sdk.serializer import serialize
    from sentry_sdk.sessions import SessionFlusher
    from sentry_sdk.traces import SpanStatus, StreamedSpan
    from sentry_sdk.tracing import trace
    from sentry_sdk.tracing_utils import has_span_streaming_enabled
    from sentry_sdk.transport import (
        AsyncHttpTransport,
        HttpTransportCore,
        make_transport,
    )
    from sentry_sdk.utils import (
        AnnotatedValue,
        ContextVar,
        capture_internal_exceptions,
        current_stacktrace,
        datetime_from_isoformat,
        env_to_bool,
        format_timestamp,
        get_before_send_log,
        get_before_send_metric,
        get_before_send_span,
        get_default_release,
        get_sdk_name,
        get_type_name,
        handle_in_app,
        has_logs_enabled,
        has_metrics_enabled,
        logger,
    )

    if TYPE_CHECKING:
        from typing import Any, Callable, Optional, Sequence, Type, TypeVar, Union

        from sentry_sdk._log_batcher import LogBatcher
        from sentry_sdk._metrics_batcher import MetricsBatcher
        from sentry_sdk._types import (
            Event,
            EventDataCategory,
            Hint,
            Log,
            Metric,
            SDKInfo,
            SerializedAttributeValue,
        )
        from sentry_sdk.integrations import Integration
        from sentry_sdk.scope import Scope
        from sentry_sdk.session import Session
        from sentry_sdk.spotlight import SpotlightClient
        from sentry_sdk.traces import StreamedSpan
        from sentry_sdk.transport import Item, Transport
        from sentry_sdk.utils import Dsn

        I = TypeVar("I", bound=Integration)  # noqa: E741

    _client_init_debug = ContextVar("client_init_debug")

    SDK_INFO: "SDKInfo" = {
        "name": "sentry.python",  # SDK name will be overridden after integrations have been loaded with sentry_sdk.integrations.setup_integrations()
        "version": VERSION,
        "packages": [{"name": "pypi:sentry-sdk", "version": VERSION}],
    }

    def _serialized_v1_attribute_to_serialized_v2_attribute(
        attribute_value: "Any",
    ) -> "Optional[SerializedAttributeValue]":
        if isinstance(attribute_value, bool):
            return {
                "value": attribute_value,
                "type": "boolean",
            }

        if isinstance(attribute_value, int):
            return {
                "value": attribute_value,
                "type": "integer",
            }

        if isinstance(attribute_value, float):
            return {
                "value": attribute_value,
                "type": "double",
            }

        if isinstance(attribute_value, str):
            return {
                "value": attribute_value,
                "type": "string",
            }

        if isinstance(attribute_value, list):
            if not attribute_value:
                return {"value": [], "type": "array"}

            ty = type(attribute_value[0])
            if ty in (int, str, bool, float) and all(
                type(v) is ty for v in attribute_value
            ):
                return {
                    "value": attribute_value,
                    "type": "array",
                }

        # Types returned when the serializer for V1 span attributes recurses into some container types.
        if isinstance(attribute_value, (dict, list)):
            return {
                "value": json.dumps(attribute_value),
                "type": "string",
            }

        return None

    def _serialized_v1_span_to_serialized_v2_span(
        span: "dict[str, Any]", event: "Event"
    ) -> "dict[str, Any]":
        # See SpanBatcher._to_transport_format() for analogous population of all entries except "attributes".
        res: "dict[str, Any]" = {
            "status": SpanStatus.OK.value,
            "is_segment": False,
        }

        if "trace_id" in span:
            res["trace_id"] = span["trace_id"]

        if "span_id" in span:
            res["span_id"] = span["span_id"]

        if "description" in span:
            description = span["description"]

            if description is None and "op" in span:
                description = span["op"]

            res["name"] = description

        if "start_timestamp" in span:
            start_timestamp = None
            try:
                start_timestamp = datetime_from_isoformat(span["start_timestamp"])
            except Exception:
                pass

            if start_timestamp is not None:
                res["start_timestamp"] = start_timestamp.timestamp()

        if "timestamp" in span:
            end_timestamp = None
            try:
                end_timestamp = datetime_from_isoformat(span["timestamp"])
            except Exception:
                pass

            if end_timestamp is not None:
                res["end_timestamp"] = end_timestamp.timestamp()

        if "parent_span_id" in span:
            res["parent_span_id"] = span["parent_span_id"]

        if "status" in span and span["status"] != SPANSTATUS.OK:
            res["status"] = "error"

        attributes: "Dict[str, Any]" = {}

        if "op" in span:
            attributes["sentry.op"] = span["op"]
        if "origin" in span:
            attributes["sentry.origin"] = span["origin"]

        span_data = span.get("data")
        if isinstance(span_data, dict):
            attributes.update(span_data)

        span_tags = span.get("tags")
        if isinstance(span_tags, dict):
            attributes.update(span_tags)

        # See Scope._apply_user_attributes_to_telemetry() for user attributes.
        user = event.get("user")
        if isinstance(user, dict):
            if "id" in user:
                attributes["user.id"] = user["id"]
            if "username" in user:
                attributes["user.name"] = user["username"]
            if "email" in user:
                attributes["user.email"] = user["email"]

        # See Scope.set_global_attributes() for release, environment, and SDK metadata.
        if "release" in event:
            attributes["sentry.release"] = event["release"]
        if "environment" in event:
            attributes["sentry.environment"] = event["environment"]
        if "server_name" in event:
            attributes["server.address"] = event["server_name"]
        if "transaction" in event:
            attributes["sentry.segment.name"] = event["transaction"]

        trace_context = event.get("contexts", {}).get("trace", {})
        if "span_id" in trace_context:
            attributes["sentry.segment.id"] = trace_context["span_id"]

        sdk_info = event.get("sdk")
        if isinstance(sdk_info, dict):
            if "name" in sdk_info:
                attributes["sentry.sdk.name"] = sdk_info["name"]
            if "version" in sdk_info:
                attributes["sentry.sdk.version"] = sdk_info["version"]

        attributes["process.runtime.name"] = platform.python_implementation()
        attributes["process.runtime.version"] = (
            f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"
        )

        if not attributes:
            return res

        res["attributes"] = {}
        for key, value in attributes.items():
            converted_value = _serialized_v1_attribute_to_serialized_v2_attribute(value)
            if converted_value is None:
                continue

            res["attributes"][key] = converted_value

        # Remove redundant attribute, as status is stored in the status field.
        if "status" in res["attributes"]:
            del res["attributes"]["status"]

        return res

    def _split_gen_ai_spans(
        event_opt: "Event",
    ) -> "Optional[tuple[List[Dict[str, object]], List[Dict[str, object]]]]":
        if "spans" not in event_opt:
            return None

        spans: "Any" = event_opt["spans"]
        if isinstance(spans, AnnotatedValue):
            spans = spans.value

        if not isinstance(spans, Iterable):
            return None

        non_gen_ai_spans = []
        gen_ai_spans = []
        for span in spans:
            if not isinstance(span, dict):
                non_gen_ai_spans.append(span)
                continue

            span_op = span.get("op")
            if isinstance(span_op, str) and span_op.startswith("gen_ai."):
                gen_ai_spans.append(span)
            else:
                non_gen_ai_spans.append(span)

        return non_gen_ai_spans, gen_ai_spans

    def _get_options(*args: "Optional[str]", **kwargs: "Any") -> "Dict[str, Any]":
        if args and (isinstance(args[0], (bytes, str)) or args[0] is None):
            dsn: "Optional[str]" = args[0]
            args = args[1:]
        else:
            dsn = None

        if len(args) > 1:
            raise TypeError("Only single positional argument is expected")

        rv = dict(DEFAULT_OPTIONS)
        options = dict(*args, **kwargs)
        if dsn is not None and options.get("dsn") is None:
            options["dsn"] = dsn

        for key, value in options.items():
            if key not in rv:
                raise TypeError("Unknown option %r" % (key,))

            rv[key] = value

        if rv["dsn"] is None:
            rv["dsn"] = os.environ.get("SENTRY_DSN")

        if rv["release"] is None:
            rv["release"] = get_default_release()

        if rv["environment"] is None:
            rv["environment"] = os.environ.get("SENTRY_ENVIRONMENT") or "production"

        if rv["debug"] is None:
            rv["debug"] = env_to_bool(os.environ.get("SENTRY_DEBUG"), strict=True) or False

        if rv["server_name"] is None and hasattr(socket, "gethostname"):
            rv["server_name"] = socket.gethostname()

        if rv["instrumenter"] is None:
            rv["instrumenter"] = INSTRUMENTER.SENTRY

        if rv["project_root"] is None:
            try:
                project_root = os.getcwd()
            except Exception:
                project_root = None

            rv["project_root"] = project_root

        if rv["enable_tracing"] is True and rv["traces_sample_rate"] is None:
            rv["traces_sample_rate"] = 1.0

        if rv["event_scrubber"] is None:
            rv["event_scrubber"] = EventScrubber(
                send_default_pii=(
                    False if rv["send_default_pii"] is None else rv["send_default_pii"]
                )
            )

        if rv["socket_options"] and not isinstance(rv["socket_options"], list):
            logger.warning(
                "Ignoring socket_options because of unexpected format. See urllib3.HTTPConnection.socket_options for the expected format."
            )
            rv["socket_options"] = None

        if rv["keep_alive"] is None:
            rv["keep_alive"] = (
                env_to_bool(os.environ.get("SENTRY_KEEP_ALIVE"), strict=True) or False
            )

        if rv["enable_tracing"] is not None:
            warnings.warn(
                "The `enable_tracing` parameter is deprecated. Please use `traces_sample_rate` instead.",
                DeprecationWarning,
                stacklevel=2,
            )

        if rv["trace_ignore_status_codes"] and has_span_streaming_enabled(rv):
            warnings.warn(
                "The `trace_ignore_status_codes` parameter is ignored in span streaming mode.",
                stacklevel=2,
            )

        return rv

    try:
        # Python 3.6+
        module_not_found_error = ModuleNotFoundError
    except Exception:
        # Older Python versions
        module_not_found_error = ImportError  # type: ignore

    class BaseClient:
        """
        .. versionadded:: 2.0.0

        The basic definition of a client that is used for sending data to Sentry.
        """

        spotlight: "Optional[SpotlightClient]" = None

        def __init__(self, options: "Optional[Dict[str, Any]]" = None) -> None:
            self.options: "Dict[str, Any]" = (
                options if options is not None else DEFAULT_OPTIONS
            )

            self.transport: "Optional[Transport]" = None
            self.monitor: "Optional[Monitor]" = None
            self.log_batcher: "Optional[LogBatcher]" = None
            self.metrics_batcher: "Optional[MetricsBatcher]" = None
            self.span_batcher: "Optional[SpanBatcher]" = None
            self.integrations: "dict[str, Integration]" = {}

        def __getstate__(self, *args: "Any", **kwargs: "Any") -> "Any":
            return {"options": {}}

        def __setstate__(self, *args: "Any", **kwargs: "Any") -> None:
            pass

        @property
        def dsn(self) -> "Optional[str]":
            return None

        @property
        def parsed_dsn(self) -> "Optional[Dsn]":
            return None

        def should_send_default_pii(self) -> bool:
            return False

        def is_active(self) -> bool:
            """
            .. versionadded:: 2.0.0

            Returns whether the client is active (able to send data to Sentry)
            """
            return False

        def capture_event(self, *args: "Any", **kwargs: "Any") -> "Optional[str]":
            return None

        def _capture_log(self, log: "Log", scope: "Scope") -> None:
            pass

        def _capture_metric(self, metric: "Metric", scope: "Scope") -> None:
            pass

        def _capture_span(self, span: "StreamedSpan", scope: "Scope") -> None:
            pass

        def capture_session(self, *args: "Any", **kwargs: "Any") -> None:
            return None

        if TYPE_CHECKING:

            @overload
            def get_integration(self, name_or_class: str) -> "Optional[Integration]": ...

            @overload
            def get_integration(self, name_or_class: "type[I]") -> "Optional[I]": ...

        def get_integration(
            self, name_or_class: "Union[str, type[Integration]]"
        ) -> "Optional[Integration]":
            return None

        def close(self, *args: "Any", **kwargs: "Any") -> None:
            return None

        def flush(self, *args: "Any", **kwargs: "Any") -> None:
            return None

        async def close_async(self, *args: "Any", **kwargs: "Any") -> None:
            return None

        async def flush_async(self, *args: "Any", **kwargs: "Any") -> None:
            return None

        def __enter__(self) -> "BaseClient":
            return self

        def __exit__(self, exc_type: "Any", exc_value: "Any", tb: "Any") -> None:
            return None

    class NonRecordingClient(BaseClient):
        """
        .. versionadded:: 2.0.0

        A client that does not send any events to Sentry. This is used as a fallback when the Sentry SDK is not yet initialized.
        """

        pass

    class _Client(BaseClient):
        """
        The client is internally responsible for capturing the events and
        forwarding them to sentry through the configured transport.  It takes
        the client options as keyword arguments and optionally the DSN as first
        argument.

        Alias of :py:class:`sentry_sdk.Client`. (Was created for better intelisense support)
        """

        def __init__(self, *args: "Any", **kwargs: "Any") -> None:
            super(_Client, self).__init__(options=get_options(*args, **kwargs))
            self._init_impl()

        def __getstate__(self) -> "Any":
            return {"options": self.options}

        def __setstate__(self, state: "Any") -> None:
            self.options = state["options"]
            self._init_impl()

        def _setup_instrumentation(
            self, functions_to_trace: "Sequence[Dict[str, str]]"
        ) -> None:
            """
            Instruments the functions given in the list `functions_to_trace` with the `@sentry_sdk.tracing.trace` decorator.
            """
            for function in functions_to_trace:
                class_name = None
                function_qualname = function["qualified_name"]

                if "." not in function_qualname:
                    logger.warning(
                        "Can not enable tracing for '%s'. Please provide the fully qualified name including the module (e.g. 'mymodule.my_function').",
                        function_qualname,
                    )
                    continue

                module_name, function_name = function_qualname.rsplit(".", 1)

                try:
                    # Try to import module and function
                    # ex: "mymodule.submodule.funcname"

                    module_obj = import_module(module_name)
                    function_obj = getattr(module_obj, function_name)
                    setattr(module_obj, function_name, trace(function_obj))
                    logger.debug("Enabled tracing for %s", function_qualname)
                except module_not_found_error:
                    try:
                        # Try to import a class
                        # ex: "mymodule.submodule.MyClassName.member_function"

                        module_name, class_name = module_name.rsplit(".", 1)
                        module_obj = import_module(module_name)
                        class_obj = getattr(module_obj, class_name)
                        function_obj = getattr(class_obj, function_name)
                        function_type = type(class_obj.__dict__[function_name])
                        traced_function = trace(function_obj)

                        if function_type in (staticmethod, classmethod):
                            traced_function = staticmethod(traced_function)

                        setattr(class_obj, function_name, traced_function)
                        setattr(module_obj, class_name, class_obj)
                        logger.debug("Enabled tracing for %s", function_qualname)

                    except Exception as e:
                        logger.warning(
                            "Can not enable tracing for '%s'. (%s) Please check your `functions_to_trace` parameter.",
                            function_qualname,
                            e,
                        )

                except Exception as e:
                    logger.warning(
                        "Can not enable tracing for '%s'. (%s) Please check your `functions_to_trace` parameter.",
                        function_qualname,
                        e,
                    )

        def _init_impl(self) -> None:
            old_debug = _client_init_debug.get(False)

            def _capture_envelope(envelope: "Envelope") -> None:
                if self.spotlight is not None:
                    self.spotlight.capture_envelope(envelope)
                if self.transport is not None:
                    self.transport.capture_envelope(envelope)

            def _record_lost_event(
                reason: str,
                data_category: "EventDataCategory",
                item: "Optional[Item]" = None,
                quantity: int = 1,
            ) -> None:
                if self.transport is not None:
                    self.transport.record_lost_event(
                        reason=reason,
                        data_category=data_category,
                        item=item,
                        quantity=quantity,
                    )

            try:
                _client_init_debug.set(self.options["debug"])
                self.transport = make_transport(self.options)

                self.monitor = None
                if self.transport:
                    if self.options["enable_backpressure_handling"]:
                        self.monitor = Monitor(self.transport)

                # Setup Spotlight before creating batchers so _capture_envelope can use it.
                # setup_spotlight handles all config/env var resolution per the SDK spec.
                from sentry_sdk.spotlight import setup_spotlight

                self.spotlight = setup_spotlight(self.options)
                if self.spotlight is not None and not self.options["dsn"]:
                    sample_all = lambda *_args, **_kwargs: 1.0
                    self.options["send_default_pii"] = True
                    self.options["error_sampler"] = sample_all
                    self.options["traces_sampler"] = sample_all
                    self.options["profiles_sampler"] = sample_all

                self.session_flusher = SessionFlusher(capture_func=_capture_envelope)

                self.log_batcher = None

                if has_logs_enabled(self.options):
                    from sentry_sdk._log_batcher import LogBatcher

                    self.log_batcher = LogBatcher(
                        capture_func=_capture_envelope,
                        record_lost_func=_record_lost_event,
                    )

                self.metrics_batcher = None
                if has_metrics_enabled(self.options):
                    self.metrics_batcher = MetricsBatcher(
                        capture_func=_capture_envelope,
                        record_lost_func=_record_lost_event,
                    )

                self.span_batcher = None
                if has_span_streaming_enabled(self.options):
                    self.span_batcher = SpanBatcher(
                        capture_func=_capture_envelope,
                        record_lost_func=_record_lost_event,
                    )

                max_request_body_size = ("always", "never", "small", "medium")
                if self.options["max_request_body_size"] not in max_request_body_size:
                    raise ValueError(
                        "Invalid value for max_request_body_size. Must be one of {}".format(
                            max_request_body_size
                        )
                    )

                if self.options["_experiments"].get("otel_powered_performance", False):
                    logger.debug(
                        "[OTel] Enabling experimental OTel-powered performance monitoring."
                    )
                    self.options["instrumenter"] = INSTRUMENTER.OTEL
                    if (
                        "sentry_sdk.integrations.opentelemetry.integration.OpenTelemetryIntegration"
                        not in _DEFAULT_INTEGRATIONS
                    ):
                        _DEFAULT_INTEGRATIONS.append(
                            "sentry_sdk.integrations.opentelemetry.integration.OpenTelemetryIntegration",
                        )

                self.integrations = setup_integrations(
                    self.options["integrations"],
                    with_defaults=self.options["default_integrations"],
                    with_auto_enabling_integrations=self.options[
                        "auto_enabling_integrations"
                    ],
                    disabled_integrations=self.options["disabled_integrations"],
                    options=self.options,
                )

                sdk_name = get_sdk_name(list(self.integrations.keys()))
                SDK_INFO["name"] = sdk_name
                logger.debug("Setting SDK name to '%s'", sdk_name)

                if has_profiling_enabled(self.options):
                    try:
                        setup_profiler(self.options)
                    except Exception as e:
                        logger.debug("Can not set up profiler. (%s)", e)
                else:
                    try:
                        setup_continuous_profiler(
                            self.options,
                            sdk_info=SDK_INFO,
                            capture_func=_capture_envelope,
                        )
                    except Exception as e:
                        logger.debug("Can not set up continuous profiler. (%s)", e)

            finally:
                _client_init_debug.set(old_debug)

            self._setup_instrumentation(self.options.get("functions_to_trace", []))

            if (
                self.monitor
                or self.log_batcher
                or self.metrics_batcher
                or self.span_batcher
                or has_profiling_enabled(self.options)
                or isinstance(self.transport, HttpTransportCore)
            ):
                # If we have anything on that could spawn a background thread, we
                # need to check if it's safe to use them.
                check_uwsgi_thread_support()

        def is_active(self) -> bool:
            """
            .. versionadded:: 2.0.0

            Returns whether the client is active (able to send data to Sentry)
            """
            return True

        def should_send_default_pii(self) -> bool:
            """
            .. versionadded:: 2.0.0

            Returns whether the client should send default PII (Personally Identifiable Information) data to Sentry.
            """
            return self.options.get("send_default_pii") or False

        @property
        def dsn(self) -> "Optional[str]":
            """Returns the configured DSN as string."""
            return self.options["dsn"]

        @property
        def parsed_dsn(self) -> "Optional[Dsn]":
            """Returns the configured parsed DSN object."""
            return self.transport.parsed_dsn if self.transport else None

        def _prepare_event(
            self,
            event: "Event",
            hint: "Hint",
            scope: "Optional[Scope]",
        ) -> "Optional[Event]":
            previous_total_spans: "Optional[int]" = None
            previous_total_breadcrumbs: "Optional[int]" = None

            if event.get("timestamp") is None:
                event["timestamp"] = datetime.now(timezone.utc)

            is_transaction = event.get("type") == "transaction"

            if scope is not None:
                spans_before = len(cast(List[Dict[str, object]], event.get("spans", [])))
                event_ = scope.apply_to_event(event, hint, self.options)

                # one of the event/error processors returned None
                if event_ is None:
                    if self.transport:
                        self.transport.record_lost_event(
                            "event_processor",
                            data_category=("transaction" if is_transaction else "error"),
                        )
                        if is_transaction:
                            self.transport.record_lost_event(
                                "event_processor",
                                data_category="span",
                                quantity=spans_before + 1,  # +1 for the transaction itself
                            )
                    return None

                event = event_
                spans_delta = spans_before - len(
                    cast(List[Dict[str, object]], event.get("spans", []))
                )
                span_recorder_dropped_spans: int = event.pop("_dropped_spans", 0)

                if is_transaction and self.transport is not None:
                    if spans_delta > 0:
                        self.transport.record_lost_event(
                            "event_processor", data_category="span", quantity=spans_delta
                        )
                    if span_recorder_dropped_spans > 0:
                        self.transport.record_lost_event(
                            "buffer_overflow",
                            data_category="span",
                            quantity=span_recorder_dropped_spans,
                        )

                dropped_spans: int = span_recorder_dropped_spans + spans_delta
                if dropped_spans > 0:
                    previous_total_spans = spans_before + dropped_spans
                if scope._n_breadcrumbs_truncated > 0:
                    breadcrumbs = event.get("breadcrumbs", {})
                    values = (
                        breadcrumbs.get("values", [])
                        if not isinstance(breadcrumbs, AnnotatedValue)
                        else []
                    )
                    previous_total_breadcrumbs = (
                        len(values) + scope._n_breadcrumbs_truncated
                    )

            if (
                not is_transaction
                and self.options["attach_stacktrace"]
                and "exception" not in event
                and "stacktrace" not in event
                and "threads" not in event
            ):
                with capture_internal_exceptions():
                    event["threads"] = {
                        "values": [
                            {
                                "stacktrace": current_stacktrace(
                                    include_local_variables=self.options.get(
                                        "include_local_variables", True
                                    ),
                                    max_value_length=self.options.get(
                                        "max_value_length", DEFAULT_MAX_VALUE_LENGTH
                                    ),
                                ),
                                "crashed": False,
                                "current": True,
                            }
                        ]
                    }

            for key in "release", "environment", "server_name", "dist":
                if event.get(key) is None and self.options[key] is not None:
                    event[key] = str(self.options[key]).strip()
            if event.get("sdk") is None:
                sdk_info = dict(SDK_INFO)
                sdk_info["integrations"] = sorted(self.integrations.keys())
                event["sdk"] = sdk_info

            if event.get("platform") is None:
                event["platform"] = "python"

            event = handle_in_app(
                event,
                self.options["in_app_exclude"],
                self.options["in_app_include"],
                self.options["project_root"],
            )

            if event is not None:
                event_scrubber = self.options["event_scrubber"]
                if event_scrubber:
                    event_scrubber.scrub_event(event)

            if scope is not None and scope._gen_ai_original_message_count:
                spans: "List[Dict[str, Any]] | AnnotatedValue" = event.get("spans", [])
                if isinstance(spans, list):
                    for span in spans:
                        span_id = span.get("span_id", None)
                        span_data = span.get("data", {})
                        if (
                            span_id
                            and span_id in scope._gen_ai_original_message_count
                            and SPANDATA.GEN_AI_REQUEST_MESSAGES in span_data
                        ):
                            span_data[SPANDATA.GEN_AI_REQUEST_MESSAGES] = AnnotatedValue(
                                span_data[SPANDATA.GEN_AI_REQUEST_MESSAGES],
                                {"len": scope._gen_ai_original_message_count[span_id]},
                            )
            if previous_total_spans is not None:
                event["spans"] = AnnotatedValue(
                    event.get("spans", []), {"len": previous_total_spans}
                )
            if previous_total_breadcrumbs is not None:
                event["breadcrumbs"] = AnnotatedValue(
                    event.get("breadcrumbs", {"values": []}),
                    {"len": previous_total_breadcrumbs},
                )

            # Postprocess the event here so that annotated types do
            # generally not surface in before_send
            if event is not None:
                event = cast(
                    "Event",
                    serialize(
                        cast("Dict[str, Any]", event),
                        max_request_body_size=self.options.get("max_request_body_size"),
                        max_value_length=self.options.get("max_value_length"),
                        custom_repr=self.options.get("custom_repr"),
                    ),
                )

            before_send = self.options["before_send"]
            if (
                before_send is not None
                and event is not None
                and event.get("type") != "transaction"
            ):
                new_event = None
                with capture_internal_exceptions():
                    new_event = before_send(event, hint or {})
                if new_event is None:
                    logger.info("before send dropped event")
                    if self.transport:
                        self.transport.record_lost_event(
                            "before_send", data_category="error"
                        )

                    # If this is an exception, reset the DedupeIntegration. It still
                    # remembers the dropped exception as the last exception, meaning
                    # that if the same exception happens again and is not dropped
                    # in before_send, it'd get dropped by DedupeIntegration.
                    if event.get("exception"):
                        DedupeIntegration.reset_last_seen()

                event = new_event

            before_send_transaction = self.options["before_send_transaction"]
            if (
                before_send_transaction is not None
                and event is not None
                and event.get("type") == "transaction"
            ):
                new_event = None
                spans_before = len(cast(List[Dict[str, object]], event.get("spans", [])))
                with capture_internal_exceptions():
                    new_event = before_send_transaction(event, hint or {})
                if new_event is None:
                    logger.info("before send transaction dropped event")
                    if self.transport:
                        self.transport.record_lost_event(
                            reason="before_send", data_category="transaction"
                        )
                        self.transport.record_lost_event(
                            reason="before_send",
                            data_category="span",
                            quantity=spans_before + 1,  # +1 for the transaction itself
                        )
                else:
                    spans_delta = spans_before - len(new_event.get("spans", []))
                    if spans_delta > 0 and self.transport is not None:
                        self.transport.record_lost_event(
                            reason="before_send", data_category="span", quantity=spans_delta
                        )

                event = new_event

            return event

        def _is_ignored_error(self, event: "Event", hint: "Hint") -> bool:
            exc_info = hint.get("exc_info")
            if exc_info is None:
                return False

            error = exc_info[0]
            error_type_name = get_type_name(exc_info[0])
            error_full_name = "%s.%s" % (exc_info[0].__module__, error_type_name)

            for ignored_error in self.options["ignore_errors"]:
                # String types are matched against the type name in the
                # exception only
                if isinstance(ignored_error, str):
                    if ignored_error == error_full_name or ignored_error == error_type_name:
                        return True
                else:
                    if issubclass(error, ignored_error):
                        return True

            return False

        def _should_capture(
            self,
            event: "Event",
            hint: "Hint",
            scope: "Optional[Scope]" = None,
        ) -> bool:
            # Transactions are sampled independent of error events.
            is_transaction = event.get("type") == "transaction"
            if is_transaction:
                return True

            ignoring_prevents_recursion = scope is not None and not scope._should_capture
            if ignoring_prevents_recursion:
                return False

            ignored_by_config_option = self._is_ignored_error(event, hint)
            if ignored_by_config_option:
                return False

            return True

        def _should_sample_error(
            self,
            event: "Event",
            hint: "Hint",
        ) -> bool:
            error_sampler = self.options.get("error_sampler", None)

            if callable(error_sampler):
                with capture_internal_exceptions():
                    sample_rate = error_sampler(event, hint)
            else:
                sample_rate = self.options["sample_rate"]

            try:
                not_in_sample_rate = sample_rate < 1.0 and random.random() >= sample_rate
            except NameError:
                logger.warning(
                    "The provided error_sampler raised an error. Defaulting to sampling the event."
                )

                # If the error_sampler raised an error, we should sample the event, since the default behavior
                # (when no sample_rate or error_sampler is provided) is to sample all events.
                not_in_sample_rate = False
            except TypeError:
                parameter, verb = (
                    ("error_sampler", "returned")
                    if callable(error_sampler)
                    else ("sample_rate", "contains")
                )
                logger.warning(
                    "The provided %s %s an invalid value of %s. The value should be a float or a bool. Defaulting to sampling the event."
                    % (parameter, verb, repr(sample_rate))
                )

                # If the sample_rate has an invalid value, we should sample the event, since the default behavior
                # (when no sample_rate or error_sampler is provided) is to sample all events.
                not_in_sample_rate = False

            if not_in_sample_rate:
                # because we will not sample this event, record a "lost event".
                if self.transport:
                    self.transport.record_lost_event("sample_rate", data_category="error")

                return False

            return True

        def _update_session_from_event(
            self,
            session: "Session",
            event: "Event",
        ) -> None:
            crashed = False
            errored = False
            user_agent = None

            exceptions = (event.get("exception") or {}).get("values")
            if exceptions:
                errored = True
                for error in exceptions:
                    if isinstance(error, AnnotatedValue):
                        error = error.value or {}
                    mechanism = error.get("mechanism")
                    if isinstance(mechanism, Mapping) and mechanism.get("handled") is False:
                        crashed = True
                        break

            user = event.get("user")

            if session.user_agent is None:
                headers = (event.get("request") or {}).get("headers")
                headers_dict = headers if isinstance(headers, dict) else {}
                for k, v in headers_dict.items():
                    if k.lower() == "user-agent":
                        user_agent = v
                        break

            session.update(
                status="crashed" if crashed else None,
                user=user,
                user_agent=user_agent,
                errors=session.errors + (errored or crashed),
            )

        def capture_event(
            self,
            event: "Event",
            hint: "Optional[Hint]" = None,
            scope: "Optional[Scope]" = None,
        ) -> "Optional[str]":
            """Captures an event.

            :param event: A ready-made event that can be directly sent to Sentry.

            :param hint: Contains metadata about the event that can be read from `before_send`, such as the original exception object or a HTTP request object.

            :param scope: An optional :py:class:`sentry_sdk.Scope` to apply to events.

            :returns: An event ID. May be `None` if there is no DSN set or of if the SDK decided to discard the event for other reasons. In such situations setting `debug=True` on `init()` may help.
            """
            hint = dict(hint or ())

            if not self._should_capture(event, hint, scope):
                return None

            profile = event.pop("profile", None)

            event_id = event.get("event_id")
            if event_id is None:
                event["event_id"] = event_id = uuid.uuid4().hex

            span_recorder_has_gen_ai_span = event.pop("_has_gen_ai_span", False)
            event_opt = self._prepare_event(event, hint, scope)
            if event_opt is None:
                return None

            # whenever we capture an event we also check if the session needs
            # to be updated based on that information.
            session = scope._session if scope else None
            if session:
                self._update_session_from_event(session, event)

            is_transaction = event_opt.get("type") == "transaction"
            is_checkin = event_opt.get("type") == "check_in"

            if (
                not is_transaction
                and not is_checkin
                and not self._should_sample_error(event, hint)
            ):
                return None

            attachments = hint.get("attachments")

            trace_context = event_opt.get("contexts", {}).get("trace") or {}
            dynamic_sampling_context = trace_context.pop("dynamic_sampling_context", {})

            headers: "dict[str, object]" = {
                "event_id": event_opt["event_id"],
                "sent_at": format_timestamp(datetime.now(timezone.utc)),
            }

            if dynamic_sampling_context:
                headers["trace"] = dynamic_sampling_context

            envelope = Envelope(headers=headers)

            if is_transaction and isinstance(profile, Profile):
                envelope.add_profile(profile.to_json(event_opt, self.options))

            if is_transaction and not span_recorder_has_gen_ai_span:
                envelope.add_transaction(event_opt)
            elif is_transaction:
                split_spans = _split_gen_ai_spans(event_opt)
                if split_spans is None or not split_spans[1]:
                    envelope.add_transaction(event_opt)
                else:
                    non_gen_ai_spans, gen_ai_spans = split_spans

                    event_opt["spans"] = non_gen_ai_spans
                    envelope.add_transaction(event_opt)

                    converted_gen_ai_spans = [
                        _serialized_v1_span_to_serialized_v2_span(span, event_opt)
                        for span in gen_ai_spans
                        if isinstance(span, dict)
                    ]

                    envelope.add_item(
                        Item(
                            type=SpanBatcher.TYPE,
                            content_type=SpanBatcher.CONTENT_TYPE,
                            headers={
                                "item_count": len(converted_gen_ai_spans),
                            },
                            payload=PayloadRef(
                                json={
                                    "version": 2,
                                    "items": converted_gen_ai_spans,
                                },
                            ),
                        )
                    )

            elif is_checkin:
                envelope.add_checkin(event_opt)
            else:
                envelope.add_event(event_opt)

            for attachment in attachments or ():
                envelope.add_item(attachment.to_envelope_item())

            return_value = None
            if self.spotlight:
                self.spotlight.capture_envelope(envelope)
                return_value = event_id

            if self.transport is not None:
                self.transport.capture_envelope(envelope)
                return_value = event_id

            return return_value

        def _capture_telemetry(
            self,
            telemetry: "Optional[Union[Log, Metric, StreamedSpan]]",
            ty: str,
            scope: "Scope",
        ) -> None:
            """
            Capture attributes-based telemetry (logs, metrics, streamed spans).

            Apply any attributes set on the scope to it, and run the user's
            before_send_{telemetry} on it, if applicable.
            """
            if telemetry is None:
                return

            scope.apply_to_telemetry(telemetry)

            before_send = None

            if ty == "log":
                before_send = get_before_send_log(self.options)
                serialized = telemetry

            elif ty == "metric":
                before_send = get_before_send_metric(self.options)
                serialized = telemetry

            elif ty == "span":
                before_send = get_before_send_span(self.options)
                serialized = telemetry._to_json()  # type: ignore[union-attr]

            if before_send is not None:
                serialized = before_send(serialized, {})  # type: ignore[arg-type]

                if ty in ("log", "metric"):
                    # Logs and metrics can be dropped in their respective
                    # before_send, so if we get None, don't queue them for sending.
                    if serialized is None:
                        return

                elif ty == "span" and isinstance(telemetry, StreamedSpan):
                    # Spans can't be dropped in before_send_span by design. They can
                    # be altered though (e.g. to sanitize). Only allow changes to
                    # name and attributes.
                    if isinstance(serialized, dict) and "name" in serialized:
                        telemetry.name = serialized["name"]  # type: ignore[typeddict-item]
                        telemetry._attributes = {}
                        for k, v in (serialized.get("attributes") or {}).items():
                            telemetry.set_attribute(k, v)

                    else:
                        logger.debug(
                            "[Tracing] Invalid return value from before_send_span. Keeping original span."
                        )

                    serialized = telemetry._to_json()

            batcher = None
            if ty == "log":
                batcher = self.log_batcher

            elif ty == "metric":
                batcher = self.metrics_batcher

            elif ty == "span":
                # We need a reference to the segment span in the batcher to populate
                # the dynamic sampling context (DSC)
                serialized["_segment_span"] = telemetry._segment  # type: ignore
                batcher = self.span_batcher

            if batcher is not None:
                batcher.add(serialized)  # type: ignore

        def _capture_log(self, log: "Optional[Log]", scope: "Scope") -> None:
            self._capture_telemetry(log, "log", scope)

        def _capture_metric(self, metric: "Optional[Metric]", scope: "Scope") -> None:
            self._capture_telemetry(metric, "metric", scope)

        def _capture_span(self, span: "Optional[StreamedSpan]", scope: "Scope") -> None:
            self._capture_telemetry(span, "span", scope)

        def capture_session(
            self,
            session: "Session",
        ) -> None:
            if not session.release:
                logger.info("Discarded session update because of missing release")
            else:
                self.session_flusher.add_session(session)

        if TYPE_CHECKING:

            @overload
            def get_integration(self, name_or_class: str) -> "Optional[Integration]": ...

            @overload
            def get_integration(self, name_or_class: "type[I]") -> "Optional[I]": ...

        def get_integration(
            self,
            name_or_class: "Union[str, Type[Integration]]",
        ) -> "Optional[Integration]":
            """Returns the integration for this client by name or class.
            If the client does not have that integration then `None` is returned.
            """
            if isinstance(name_or_class, str):
                integration_name = name_or_class
            elif name_or_class.identifier is not None:
                integration_name = name_or_class.identifier
            else:
                raise ValueError("Integration has no name")

            return self.integrations.get(integration_name)

        def _has_async_transport(self) -> bool:
            """Check if the current transport is async."""
            return isinstance(self.transport, AsyncHttpTransport)

        @property
        def _batchers(self) -> "tuple[Any, ...]":
            return tuple(
                b
                for b in (self.log_batcher, self.metrics_batcher, self.span_batcher)
                if b is not None
            )

        def _close_components(self) -> None:
            """Kill all client components in the correct order."""
            self.session_flusher.kill()
            for b in self._batchers:
                b.kill()
            if self.monitor:
                self.monitor.kill()

        def _flush_components(self) -> None:
            """Flush all client components."""
            self.session_flusher.flush()
            for b in self._batchers:
                b.flush()

        def close(
            self,
            timeout: "Optional[float]" = None,
            callback: "Optional[Callable[[int, float], None]]" = None,
        ) -> None:
            """
            Close the client and shut down the transport. Arguments have the same
            semantics as :py:meth:`Client.flush`.
            """
            if self.transport is not None:
                if self._has_async_transport():
                    warnings.warn(
                        "close() used with AsyncHttpTransport. Use close_async() instead.",
                        stacklevel=2,
                    )
                    self._flush_components()
                else:
                    self.flush(timeout=timeout, callback=callback)
                self._close_components()
                self.transport.kill()
                self.transport = None

        async def close_async(
            self,
            timeout: "Optional[float]" = None,
            callback: "Optional[Callable[[int, float], None]]" = None,
        ) -> None:
            """
            Asynchronously close the client and shut down the transport. Arguments have the same
            semantics as :py:meth:`Client.flush_async`.
            """
            if self.transport is not None:
                if not self._has_async_transport():
                    logger.debug(
                        "close_async() used with non-async transport, aborting. Please use close() instead."
                    )
                    return
                await self.flush_async(timeout=timeout, callback=callback)
                self._close_components()
                kill_task = self.transport.kill()  # type: ignore
                if kill_task is not None:
                    await kill_task
                self.transport = None

        def flush(
            self,
            timeout: "Optional[float]" = None,
            callback: "Optional[Callable[[int, float], None]]" = None,
        ) -> None:
            """
            Wait for the current events to be sent.

            :param timeout: Wait for at most `timeout` seconds. If no `timeout` is provided, the `shutdown_timeout` option value is used.

            :param callback: Is invoked with the number of pending events and the configured timeout.
            """
            if self.transport is not None:
                if self._has_async_transport():
                    warnings.warn(
                        "flush() used with AsyncHttpTransport. Use flush_async() instead.",
                        stacklevel=2,
                    )
                    return
                if timeout is None:
                    timeout = self.options["shutdown_timeout"]
                self._flush_components()

                self.transport.flush(timeout=timeout, callback=callback)

        async def flush_async(
            self,
            timeout: "Optional[float]" = None,
            callback: "Optional[Callable[[int, float], None]]" = None,
        ) -> None:
            """
            Asynchronously wait for the current events to be sent.

            :param timeout: Wait for at most `timeout` seconds. If no `timeout` is provided, the `shutdown_timeout` option value is used.

            :param callback: Is invoked with the number of pending events and the configured timeout.
            """
            if self.transport is not None:
                if not self._has_async_transport():
                    logger.debug(
                        "flush_async() used with non-async transport, aborting. Please use flush() instead."
                    )
                    return
                if timeout is None:
                    timeout = self.options["shutdown_timeout"]
                self._flush_components()
                flush_task = self.transport.flush(timeout=timeout, callback=callback)  # type: ignore
                if flush_task is not None:
                    await flush_task

        def __enter__(self) -> "_Client":
            return self

        def __exit__(self, exc_type: "Any", exc_value: "Any", tb: "Any") -> None:
            self.close()

        async def __aenter__(self) -> "_Client":
            return self

        async def __aexit__(self, exc_type: "Any", exc_value: "Any", tb: "Any") -> None:
            await self.close_async()

    from typing import TYPE_CHECKING

    if TYPE_CHECKING:
        # Make mypy, PyCharm and other static analyzers think `get_options` is a
        # type to have nicer autocompletion for params.
        #
        # Use `ClientConstructor` to define the argument types of `init` and
        # `Dict[str, Any]` to tell static analyzers about the return type.

        class get_options(ClientConstructor, Dict[str, Any]):  # noqa: N801
            pass

        class Client(ClientConstructor, _Client):
            pass

    else:
        # Alias `get_options` for actual usage. Go through the lambda indirection
        # to throw PyCharm off of the weakly typed signature (it would otherwise
        # discover both the weakly typed signature of `_init` and our faked `init`
        # type).

        get_options = (lambda: _get_options)()
        Client = (lambda: _Client)()