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._log_batcher import LogBatcher

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.data_collection import (

_map_from_send_default_pii,

_resolve_data_collection,

)

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_data_collection_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

rv["data_collection"] = _resolve_data_collection(rv)

# Do not add the event scrubber if data collection is enabled as it can remove data that's

# collected under data collection config

if not has_data_collection_enabled(rv) and 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"]

)

elif has_data_collection_enabled(rv) and rv["event_scrubber"]:

warnings.warn(

"Event scrubbers are not enabled when data collection configuration is provided. Ignoring event_scrubber...",

stacklevel=2,

)

rv["event_scrubber"] = None

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,

)

if rv["ignore_spans"] and not has_span_streaming_enabled(rv):

warnings.warn(

"The `ignore_spans` parameter only works when `trace_lifecycle` is set to `stream`.",

stacklevel=2,

)

if rv["before_send_span"] and not has_span_streaming_enabled(rv):

warnings.warn(

"The `before_send_span` parameter only works when `trace_lifecycle` is set to `stream`.",

stacklevel=2,

)

return rv

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 ModuleNotFoundError:

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

# data_collection was resolved in _get_options() before this

# spotlight override flipped send_default_pii on. Re-derive it so

# data_collection agrees with should_send_default_pii() in

# DSN-less spotlight mode (only when the user did not set

# data_collection explicitly).

if not self.options["data_collection"]["provided_by_user"]:

self.options["data_collection"] = _map_from_send_default_pii(

send_default_pii=True,

include_local_variables=self.options["include_local_variables"]

is not False,

include_source_context=self.options["include_source_context"]

is not False,

)

self.session_flusher = SessionFlusher(capture_func=_capture_envelope)

if self.options.get("enable_logs", False) or self.options[

"_experiments"

].get("enable_logs", False):

logger.warning(

"The enable_logs option has no effect and will be removed in the next major."

)

self.log_batcher = LogBatcher(

capture_func=_capture_envelope,

record_lost_func=_record_lost_event,

)

if self.options.get("enable_metrics", True) is False:

logger.warning(

"The enable_metrics option has no effect and will be removed in the next major."

)

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