import base64

import copy

import json

import linecache

import logging

import math

import os

import random

import re

import subprocess

import sys

import threading

import time

from collections import namedtuple

from contextlib import contextmanager

from datetime import datetime, timezone

from decimal import Decimal

from functools import partial, partialmethod, wraps

from numbers import Real

from urllib.parse import parse_qs, unquote, urlencode, urlsplit, urlunsplit

try:

# Python 3.11

from builtins import BaseExceptionGroup

except ImportError:

# Python 3.10 and below

BaseExceptionGroup = None # type: ignore

from typing import TYPE_CHECKING

import sentry_sdk

from sentry_sdk._compat import PY37

from sentry_sdk._types import SENSITIVE_DATA_SUBSTITUTE, Annotated, AnnotatedValue

from sentry_sdk.consts import (

DEFAULT_ADD_FULL_STACK,

DEFAULT_MAX_STACK_FRAMES,

EndpointType,

)

if TYPE_CHECKING:

from types import FrameType, TracebackType

from typing import (

Any,

Callable,

ContextManager,

Dict,

Generator,

Iterator,

List,

NoReturn,

Optional,

ParamSpec,

Set,

Tuple,

Type,

TypeVar,

Union,

cast,

overload,

)

from gevent.hub import Hub

from sentry_sdk._types import (

AttributeValue,

Event,

ExcInfo,

Hint,

Log,

Metric,

SerializedAttributeValue,

SpanJSON,

)

P = ParamSpec("P")

R = TypeVar("R")

epoch = datetime(1970, 1, 1)

# The logger is created here but initialized in the debug support module

logger = logging.getLogger("sentry_sdk.errors")

_installed_modules = None

BASE64_ALPHABET = re.compile(r"^[a-zA-Z0-9/+=]*$")

FALSY_ENV_VALUES = frozenset(("false", "f", "n", "no", "off", "0"))

TRUTHY_ENV_VALUES = frozenset(("true", "t", "y", "yes", "on", "1"))

MAX_STACK_FRAMES = 2000

"""Maximum number of stack frames to send to Sentry.

If we have more than this number of stack frames, we will stop processing

the stacktrace to avoid getting stuck in a long-lasting loop. This value

exceeds the default sys.getrecursionlimit() of 1000, so users will only

be affected by this limit if they have a custom recursion limit.

"""

def env_to_bool(value: "Any", *, strict: "Optional[bool]" = False) -> "bool | None":

"""Casts an ENV variable value to boolean using the constants defined above.

In strict mode, it may return None if the value doesn't match any of the predefined values.

"""

normalized = str(value).lower() if value is not None else None

if normalized in FALSY_ENV_VALUES:

return False

if normalized in TRUTHY_ENV_VALUES:

return True

return None if strict else bool(value)

def json_dumps(data: "Any") -> bytes:

"""Serialize data into a compact JSON representation encoded as UTF-8."""

return json.dumps(data, allow_nan=False, separators=(",", ":")).encode("utf-8")

def get_git_revision() -> "Optional[str]":

try:

with open(os.path.devnull, "w+") as null:

# prevent command prompt windows from popping up on windows

startupinfo = None

if sys.platform == "win32" or sys.platform == "cygwin":

startupinfo = subprocess.STARTUPINFO()

startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW

revision = (

subprocess.Popen(

["git", "rev-parse", "HEAD"],

startupinfo=startupinfo,

stdout=subprocess.PIPE,

stderr=null,

stdin=null,

)

.communicate()[0]

.strip()

.decode("utf-8")

)

except (OSError, IOError, FileNotFoundError):

return None

return revision

def get_default_release() -> "Optional[str]":

"""Try to guess a default release."""

release = os.environ.get("SENTRY_RELEASE")

if release:

return release

release = get_git_revision()

if release:

return release

for var in (

"HEROKU_BUILD_COMMIT",

"HEROKU_SLUG_COMMIT", # deprecated by Heroku, kept for backward compatibility

"SOURCE_VERSION",

"CODEBUILD_RESOLVED_SOURCE_VERSION",

"CIRCLE_SHA1",

"GAE_DEPLOYMENT_ID",

"K_REVISION",

):

release = os.environ.get(var)

if release:

return release

return None

def get_sdk_name(installed_integrations: "List[str]") -> str:

"""Return the SDK name including the name of the used web framework."""

# Note: I can not use for example sentry_sdk.integrations.django.DjangoIntegration.identifier

# here because if django is not installed the integration is not accessible.

framework_integrations = [

"django",

"flask",

"fastapi",

"bottle",

"falcon",

"quart",

"sanic",

"starlette",

"litestar",

"starlite",

"chalice",

"serverless",

"pyramid",

"tornado",

"aiohttp",

"aws_lambda",

"gcp",

"beam",

"asgi",

"wsgi",

]

for integration in framework_integrations:

if integration in installed_integrations:

return "sentry.python.{}".format(integration)

return "sentry.python"

class CaptureInternalException:

__slots__ = ()

def __enter__(self) -> "ContextManager[Any]":

return self

def __exit__(

self,

ty: "Optional[Type[BaseException]]",

value: "Optional[BaseException]",

tb: "Optional[TracebackType]",

) -> bool:

if ty is not None and value is not None:

capture_internal_exception((ty, value, tb))

return True

_CAPTURE_INTERNAL_EXCEPTION = CaptureInternalException()

def capture_internal_exceptions() -> "ContextManager[Any]":

return _CAPTURE_INTERNAL_EXCEPTION

def capture_internal_exception(exc_info: "ExcInfo") -> None:

"""

Capture an exception that is likely caused by a bug in the SDK

itself.

These exceptions do not end up in Sentry and are just logged instead.

"""

if sentry_sdk.get_client().is_active():

logger.error("Internal error in sentry_sdk", exc_info=exc_info)

def to_timestamp(value: "datetime") -> float:

return (value - epoch).total_seconds()

def format_timestamp(value: "datetime") -> str:

"""Formats a timestamp in RFC 3339 format.

Any datetime objects with a non-UTC timezone are converted to UTC, so that all timestamps are formatted in UTC.

"""

utctime = value.astimezone(timezone.utc)

# We use this custom formatting rather than isoformat for backwards compatibility (we have used this format for

# several years now), and isoformat is slightly different.

return utctime.strftime("%Y-%m-%dT%H:%M:%S.%fZ")

ISO_TZ_SEPARATORS = frozenset(("+", "-"))

def datetime_from_isoformat(value: str) -> "datetime":

try:

result = datetime.fromisoformat(value)

except (AttributeError, ValueError):

# py 3.6

timestamp_format = (

"%Y-%m-%dT%H:%M:%S.%f" if "." in value else "%Y-%m-%dT%H:%M:%S"

)

if value.endswith("Z"):

value = value[:-1] + "+0000"

if value[-6] in ISO_TZ_SEPARATORS:

timestamp_format += "%z"

value = value[:-3] + value[-2:]

elif value[-5] in ISO_TZ_SEPARATORS:

timestamp_format += "%z"

result = datetime.strptime(value, timestamp_format)

return result.astimezone(timezone.utc)

def event_hint_with_exc_info(

exc_info: "Optional[ExcInfo]" = None,

) -> "Dict[str, Optional[ExcInfo]]":

"""Creates a hint with the exc info filled in."""

if exc_info is None:

exc_info = sys.exc_info()

else:

exc_info = exc_info_from_error(exc_info)

if exc_info[0] is None:

exc_info = None

return {"exc_info": exc_info}

class BadDsn(ValueError):

"""Raised on invalid DSNs."""

class Dsn:

"""Represents a DSN."""

ORG_ID_REGEX = re.compile(r"^o(\d+)\.")

def __init__(

self, value: "Union[Dsn, str]", org_id: "Optional[str]" = None

) -> None:

if isinstance(value, Dsn):

self.__dict__ = dict(value.__dict__)

return

parts = urlsplit(str(value))

if parts.scheme not in ("http", "https"):

raise BadDsn("Unsupported scheme %r" % parts.scheme)

self.scheme = parts.scheme

if parts.hostname is None:

raise BadDsn("Missing hostname")

self.host = parts.hostname

if org_id is not None:

self.org_id: "Optional[str]" = org_id

else:

org_id_match = Dsn.ORG_ID_REGEX.match(self.host)

self.org_id = org_id_match.group(1) if org_id_match else None

if parts.port is None:

self.port: int = self.scheme == "https" and 443 or 80

else:

self.port = parts.port

if not parts.username:

raise BadDsn("Missing public key")

self.public_key = parts.username

self.secret_key = parts.password

path = parts.path.rsplit("/", 1)

try:

self.project_id = str(int(path.pop()))

except (ValueError, TypeError):

raise BadDsn("Invalid project in DSN (%r)" % (parts.path or "")[1:])

self.path = "/".join(path) + "/"

@property

def netloc(self) -> str:

"""The netloc part of a DSN."""

rv = self.host

if (self.scheme, self.port) not in (("http", 80), ("https", 443)):

rv = "%s:%s" % (rv, self.port)

return rv

def to_auth(self, client: "Optional[Any]" = None) -> "Auth":

"""Returns the auth info object for this dsn."""

return Auth(

scheme=self.scheme,

host=self.netloc,

path=self.path,

project_id=self.project_id,

public_key=self.public_key,

secret_key=self.secret_key,

client=client,

)

def __str__(self) -> str:

return "%s://%s%s@%s%s%s" % (

self.scheme,

self.public_key,

self.secret_key and "@" + self.secret_key or "",

self.netloc,

self.path,

self.project_id,

)

class Auth:

"""Helper object that represents the auth info."""

def __init__(

self,

scheme: str,

host: str,

project_id: str,

public_key: str,

secret_key: "Optional[str]" = None,

version: int = 7,

client: "Optional[Any]" = None,

path: str = "/",

) -> None:

self.scheme = scheme

self.host = host

self.path = path

self.project_id = project_id

self.public_key = public_key

self.secret_key = secret_key

self.version = version

self.client = client

def get_api_url(

self,

type: "EndpointType" = EndpointType.ENVELOPE,

) -> str:

"""Returns the API url for storing events."""

return "%s://%s%sapi/%s/%s/" % (

self.scheme,

self.host,

self.path,

self.project_id,

type.value,

)

def to_header(self) -> str:

"""Returns the auth header a string."""

rv = [("sentry_key", self.public_key), ("sentry_version", self.version)]

if self.client is not None:

rv.append(("sentry_client", self.client))

if self.secret_key is not None:

rv.append(("sentry_secret", self.secret_key))

return "Sentry " + ", ".join("%s=%s" % (key, value) for key, value in rv)

def get_type_name(cls: "Optional[type]") -> "Optional[str]":

return getattr(cls, "__qualname__", None) or getattr(cls, "__name__", None)

def get_type_module(cls: "Optional[type]") -> "Optional[str]":

mod = getattr(cls, "__module__", None)

if mod not in (None, "builtins", "__builtins__"):

return mod

return None

def should_hide_frame(frame: "FrameType") -> bool:

try:

mod = frame.f_globals["__name__"]

if mod.startswith("sentry_sdk."):

return True

except (AttributeError, KeyError):

pass

for flag_name in "__traceback_hide__", "__tracebackhide__":

try:

if frame.f_locals[flag_name]:

return True

except Exception:

pass

return False

def iter_stacks(tb: "Optional[TracebackType]") -> "Iterator[TracebackType]":

tb_: "Optional[TracebackType]" = tb

while tb_ is not None:

if not should_hide_frame(tb_.tb_frame):

yield tb_

tb_ = tb_.tb_next

def get_lines_from_file(

filename: str,

lineno: int,

max_length: "Optional[int]" = None,

loader: "Optional[Any]" = None,

module: "Optional[str]" = None,

) -> "Tuple[List[Annotated[str]], Optional[Annotated[str]], List[Annotated[str]]]":

context_lines = 5

source = None

if loader is not None and hasattr(loader, "get_source"):

try:

source_str: "Optional[str]" = loader.get_source(module)

except (ImportError, IOError):

source_str = None

if source_str is not None:

source = source_str.splitlines()

if source is None:

try:

source = linecache.getlines(filename)

except (OSError, IOError):

return [], None, []

if not source:

return [], None, []

lower_bound = max(0, lineno - context_lines)

upper_bound = min(lineno + 1 + context_lines, len(source))

try:

pre_context = [

strip_string(line.strip("\r\n"), max_length=max_length)

for line in source[lower_bound:lineno]

]

context_line = strip_string(source[lineno].strip("\r\n"), max_length=max_length)

post_context = [

strip_string(line.strip("\r\n"), max_length=max_length)

for line in source[(lineno + 1) : upper_bound]

]

return pre_context, context_line, post_context

except IndexError:

# the file may have changed since it was loaded into memory

return [], None, []

def get_source_context(

frame: "FrameType",

tb_lineno: "Optional[int]",

max_value_length: "Optional[int]" = None,

) -> "Tuple[List[Annotated[str]], Optional[Annotated[str]], List[Annotated[str]]]":

try:

abs_path: "Optional[str]" = frame.f_code.co_filename

except Exception:

abs_path = None

try:

module = frame.f_globals["__name__"]

except Exception:

return [], None, []

try:

loader = frame.f_globals["__loader__"]

except Exception:

loader = None

if tb_lineno is not None and abs_path:

lineno = tb_lineno - 1

return get_lines_from_file(

abs_path, lineno, max_value_length, loader=loader, module=module

)

return [], None, []

def safe_str(value: "Any") -> str:

try:

return str(value)

except Exception:

return safe_repr(value)

def safe_repr(value: "Any") -> str:

try:

return repr(value)

except Exception:

return "<broken repr>"

def filename_for_module(

module: "Optional[str]", abs_path: "Optional[str]"

) -> "Optional[str]":

if not abs_path or not module:

return abs_path

try:

if abs_path.endswith(".pyc"):

abs_path = abs_path[:-1]

base_module = module.split(".", 1)[0]

if base_module == module:

return os.path.basename(abs_path)

base_module_path = sys.modules[base_module].__file__

if not base_module_path:

return abs_path

return abs_path.split(base_module_path.rsplit(os.sep, 2)[0], 1)[-1].lstrip(

os.sep

)

except Exception:

return abs_path

def serialize_frame(

frame: "FrameType",

tb_lineno: "Optional[int]" = None,

include_local_variables: bool = True,

include_source_context: bool = True,

max_value_length: "Optional[int]" = None,

custom_repr: "Optional[Callable[..., Optional[str]]]" = None,

) -> "Dict[str, Any]":

f_code = getattr(frame, "f_code", None)

if not f_code:

abs_path = None

function = None

else:

abs_path = frame.f_code.co_filename

function = frame.f_code.co_name

try:

module = frame.f_globals["__name__"]

except Exception:

module = None

if tb_lineno is None:

tb_lineno = frame.f_lineno

try:

os_abs_path = os.path.abspath(abs_path) if abs_path else None

except Exception:

os_abs_path = None

rv: "Dict[str, Any]" = {

"filename": filename_for_module(module, abs_path) or None,

"abs_path": os_abs_path,

"function": function or "<unknown>",

"module": module,

"lineno": tb_lineno,

}

if include_source_context:

rv["pre_context"], rv["context_line"], rv["post_context"] = get_source_context(

frame, tb_lineno, max_value_length

)

if include_local_variables:

from sentry_sdk.serializer import serialize

rv["vars"] = serialize(

dict(frame.f_locals), is_vars=True, custom_repr=custom_repr

)

return rv

def current_stacktrace(

include_local_variables: bool = True,

include_source_context: bool = True,

max_value_length: "Optional[int]" = None,

) -> "Dict[str, Any]":

__tracebackhide__ = True

frames = []

f: "Optional[FrameType]" = sys._getframe()

while f is not None:

if not should_hide_frame(f):

frames.append(

serialize_frame(

f,

include_local_variables=include_local_variables,

include_source_context=include_source_context,

max_value_length=max_value_length,

)

)

f = f.f_back

frames.reverse()

return {"frames": frames}

def get_errno(exc_value: BaseException) -> "Optional[Any]":

return getattr(exc_value, "errno", None)

def get_error_message(exc_value: "Optional[BaseException]") -> str:

message: str = safe_str(

getattr(exc_value, "message", "")

or getattr(exc_value, "detail", "")

or safe_str(exc_value)

)

# __notes__ should be a list of strings when notes are added

# via add_note, but can be anything else if __notes__ is set

# directly. We only support strings in __notes__, since that

# is the correct use.

notes: object = getattr(exc_value, "__notes__", None)

if isinstance(notes, list) and len(notes) > 0:

message += "\n" + "\n".join(note for note in notes if isinstance(note, str))

return message

def single_exception_from_error_tuple(

exc_type: "Optional[type]",

exc_value: "Optional[BaseException]",

tb: "Optional[TracebackType]",

client_options: "Optional[Dict[str, Any]]" = None,

mechanism: "Optional[Dict[str, Any]]" = None,

exception_id: "Optional[int]" = None,

parent_id: "Optional[int]" = None,

source: "Optional[str]" = None,

full_stack: "Optional[list[dict[str, Any]]]" = None,

) -> "Dict[str, Any]":

"""

Creates a dict that goes into the events `exception.values` list and is ingestible by Sentry.

See the Exception Interface documentation for more details:

https://develop.sentry.dev/sdk/event-payloads/exception/

"""

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

exception_value["mechanism"] = (

mechanism.copy() if mechanism else {"type": "generic", "handled": True}

)

if exception_id is not None:

exception_value["mechanism"]["exception_id"] = exception_id

if exc_value is not None:

errno = get_errno(exc_value)

else:

errno = None

if errno is not None:

exception_value["mechanism"].setdefault("meta", {}).setdefault(

"errno", {}

).setdefault("number", errno)

if source is not None:

exception_value["mechanism"]["source"] = source

is_root_exception = exception_id == 0

if not is_root_exception and parent_id is not None:

exception_value["mechanism"]["parent_id"] = parent_id

exception_value["mechanism"]["type"] = "chained"

if is_root_exception and "type" not in exception_value["mechanism"]:

exception_value["mechanism"]["type"] = "generic"

is_exception_group = BaseExceptionGroup is not None and isinstance(

exc_value, BaseExceptionGroup

)

if is_exception_group:

exception_value["mechanism"]["is_exception_group"] = True

exception_value["module"] = get_type_module(exc_type)

exception_value["type"] = get_type_name(exc_type)

exception_value["value"] = get_error_message(exc_value)

if client_options is None:

include_local_variables = True

include_source_context = True

max_value_length = None # fallback

custom_repr = None

else:

include_local_variables = client_options["include_local_variables"]

include_source_context = client_options["include_source_context"]

max_value_length = client_options["max_value_length"]

custom_repr = client_options.get("custom_repr")

frames: "List[Dict[str, Any]]" = [

serialize_frame(

tb.tb_frame,

tb_lineno=tb.tb_lineno,

include_local_variables=include_local_variables,

include_source_context=include_source_context,

max_value_length=max_value_length,

custom_repr=custom_repr,

)

# Process at most MAX_STACK_FRAMES + 1 frames, to avoid hanging on

# processing a super-long stacktrace.

for tb, _ in zip(iter_stacks(tb), range(MAX_STACK_FRAMES + 1))

]

if len(frames) > MAX_STACK_FRAMES:

# If we have more frames than the limit, we remove the stacktrace completely.

# We don't trim the stacktrace here because we have not processed the whole

# thing (see above, we stop at MAX_STACK_FRAMES + 1). Normally, Relay would

# intelligently trim by removing frames in the middle of the stacktrace, but

# since we don't have the whole stacktrace, we can't do that. Instead, we

# drop the entire stacktrace.

exception_value["stacktrace"] = AnnotatedValue.removed_because_over_size_limit(

value=None

)

elif frames:

if not full_stack:

new_frames = frames

else:

new_frames = merge_stack_frames(frames, full_stack, client_options)

exception_value["stacktrace"] = {"frames": new_frames}

return exception_value

HAS_CHAINED_EXCEPTIONS = hasattr(Exception, "__suppress_context__")

if HAS_CHAINED_EXCEPTIONS:

def walk_exception_chain(exc_info: "ExcInfo") -> "Iterator[ExcInfo]":

exc_type, exc_value, tb = exc_info

seen_exceptions = []

seen_exception_ids: "Set[int]" = set()

while (

exc_type is not None

and exc_value is not None

and id(exc_value) not in seen_exception_ids

):

yield exc_type, exc_value, tb

# Avoid hashing random types we don't know anything

# about. Use the list to keep a ref so that the `id` is

# not used for another object.

seen_exceptions.append(exc_value)

seen_exception_ids.add(id(exc_value))

if exc_value.__suppress_context__:

cause = exc_value.__cause__

else:

cause = exc_value.__context__

if cause is None:

break

exc_type = type(cause)

exc_value = cause

tb = getattr(cause, "__traceback__", None)

else:

def walk_exception_chain(exc_info: "ExcInfo") -> "Iterator[ExcInfo]":

yield exc_info

def exceptions_from_error(

exc_type: "Optional[type]",

exc_value: "Optional[BaseException]",

tb: "Optional[TracebackType]",

client_options: "Optional[Dict[str, Any]]" = None,

mechanism: "Optional[Dict[str, Any]]" = None,

exception_id: int = 0,

parent_id: int = 0,

source: "Optional[str]" = None,

full_stack: "Optional[list[dict[str, Any]]]" = None,

seen_exceptions: "Optional[list[BaseException]]" = None,

seen_exception_ids: "Optional[Set[int]]" = None,

) -> "Tuple[int, List[Dict[str, Any]]]":

"""

Creates the list of exceptions.

This can include chained exceptions and exceptions from an ExceptionGroup.

See the Exception Interface documentation for more details:

https://develop.sentry.dev/sdk/event-payloads/exception/

Args:

exception_id (int):

Sequential counter for assigning ``mechanism.exception_id``

to each processed exception. Is NOT the result of calling `id()` on the exception itself.

parent_id (int):

The ``mechanism.exception_id`` of the parent exception.

Written into ``mechanism.parent_id`` in the event payload so Sentry can

reconstruct the exception tree.

Not to be confused with ``seen_exception_ids``, which tracks Python ``id()``

values for cycle detection.

"""

if seen_exception_ids is None:

seen_exception_ids = set()

if seen_exceptions is None:

seen_exceptions = []

if exc_value is not None and id(exc_value) in seen_exception_ids:

return (exception_id, [])

if exc_value is not None:

seen_exceptions.append(exc_value)

seen_exception_ids.add(id(exc_value))

parent = single_exception_from_error_tuple(

exc_type=exc_type,

exc_value=exc_value,

tb=tb,

client_options=client_options,

mechanism=mechanism,

exception_id=exception_id,

parent_id=parent_id,

source=source,

full_stack=full_stack,

)

exceptions = [parent]

parent_id = exception_id

exception_id += 1

should_supress_context = (

hasattr(exc_value, "__suppress_context__") and exc_value.__suppress_context__ # type: ignore

)

if should_supress_context:

# Add direct cause.

# The field `__cause__` is set when raised with the exception (using the `from` keyword).

exception_has_cause = (

exc_value

and hasattr(exc_value, "__cause__")

and exc_value.__cause__ is not None

)

if exception_has_cause:

cause = exc_value.__cause__ # type: ignore

(exception_id, child_exceptions) = exceptions_from_error(

exc_type=type(cause),

exc_value=cause,

tb=getattr(cause, "__traceback__", None),

client_options=client_options,

mechanism=mechanism,

exception_id=exception_id,

source="__cause__",

full_stack=full_stack,

seen_exceptions=seen_exceptions,

seen_exception_ids=seen_exception_ids,

)

exceptions.extend(child_exceptions)

else:

# Add indirect cause.

# The field `__context__` is assigned if another exception occurs while handling the exception.

exception_has_content = (

exc_value

and hasattr(exc_value, "__context__")

and exc_value.__context__ is not None

)

if exception_has_content:

context = exc_value.__context__ # type: ignore

(exception_id, child_exceptions) = exceptions_from_error(

exc_type=type(context),

exc_value=context,

tb=getattr(context, "__traceback__", None),

client_options=client_options,

mechanism=mechanism,

exception_id=exception_id,

source="__context__",

full_stack=full_stack,

seen_exceptions=seen_exceptions,

seen_exception_ids=seen_exception_ids,

)

exceptions.extend(child_exceptions)

# Add exceptions from an ExceptionGroup.

is_exception_group = exc_value and hasattr(exc_value, "exceptions")

if is_exception_group:

for idx, e in enumerate(exc_value.exceptions): # type: ignore

(exception_id, child_exceptions) = exceptions_from_error(

exc_type=type(e),

exc_value=e,

tb=getattr(e, "__traceback__", None),

client_options=client_options,

mechanism=mechanism,

exception_id=exception_id,

parent_id=parent_id,

source="exceptions[%s]" % idx,

full_stack=full_stack,

seen_exceptions=seen_exceptions,

seen_exception_ids=seen_exception_ids,

)

exceptions.extend(child_exceptions)

return (exception_id, exceptions)

def exceptions_from_error_tuple(

exc_info: "ExcInfo",

client_options: "Optional[Dict[str, Any]]" = None,

mechanism: "Optional[Dict[str, Any]]" = None,

full_stack: "Optional[list[dict[str, Any]]]" = None,

) -> "List[Dict[str, Any]]":

exc_type, exc_value, tb = exc_info

is_exception_group = BaseExceptionGroup is not None and isinstance(

exc_value, BaseExceptionGroup

)

if is_exception_group:

(_, exceptions) = exceptions_from_error(

exc_type=exc_type,

exc_value=exc_value,

tb=tb,

client_options=client_options,

mechanism=mechanism,

exception_id=0,

parent_id=0,

full_stack=full_stack,

)

else:

exceptions = []

for exc_type, exc_value, tb in walk_exception_chain(exc_info):

exceptions.append(

single_exception_from_error_tuple(

exc_type=exc_type,

exc_value=exc_value,

tb=tb,

client_options=client_options,

mechanism=mechanism,

full_stack=full_stack,

)

)

exceptions.reverse()

return exceptions

def to_string(value: str) -> str:

try:

return str(value)

except UnicodeDecodeError:

return repr(value)[1:-1]