# pre-release
                [-_\.]?
                (?P(a|b|c|rc|alpha|beta|pre|preview))
                [-_\.]?
                (?P[0-9]+)?
            )?
            (?P                                         # post release
                (?:-(?P[0-9]+))
                |
                (?:
                    [-_\.]?
                    (?Ppost|rev|r)
                    [-_\.]?
                    (?P[0-9]+)?
                )
            )?
            (?P                                          # dev release
                [-_\.]?
                (?Pdev)
                [-_\.]?
                (?P[0-9]+)?
            )?
        )
        (?:\+(?P[a-z0-9]+(?:[-_\.][a-z0-9]+)*))?       # local version
    """

    pattern = re.compile(
        r"^\s*" + VERSION_PATTERN + r"\s*$",
        re.VERBOSE | re.IGNORECASE,
    )

    try:
        release = pattern.match(version).groupdict()["release"]  # type: ignore
        release_tuple: "Tuple[int, ...]" = tuple(map(int, release.split(".")[:3]))
    except (TypeError, ValueError, AttributeError):
        return None

    return release_tuple


def _is_contextvars_broken() -> bool:
    """
    Returns whether gevent/eventlet have patched the stdlib in a way where thread locals are now more "correct" than contextvars.
    """
    try:
        import gevent
        from gevent.monkey import is_object_patched

        # Get the MAJOR and MINOR version numbers of Gevent
        version_tuple = tuple(
            [int(part) for part in re.split(r"a|b|rc|\.", gevent.__version__)[:2]]
        )
        if is_object_patched("threading", "local"):
            # Gevent 20.9.0 depends on Greenlet 0.4.17 which natively handles switching
            # context vars when greenlets are switched, so, Gevent 20.9.0+ is all fine.
            # Ref: https://github.com/gevent/gevent/blob/83c9e2ae5b0834b8f84233760aabe82c3ba065b4/src/gevent/monkey.py#L604-L609
            # Gevent 20.5, that doesn't depend on Greenlet 0.4.17 with native support
            # for contextvars, is able to patch both thread locals and contextvars, in
            # that case, check if contextvars are effectively patched.
            if (
                # Gevent 20.9.0+
                (sys.version_info >= (3, 7) and version_tuple >= (20, 9))
                # Gevent 20.5.0+ or Python < 3.7
                or (is_object_patched("contextvars", "ContextVar"))
            ):
                return False

            return True
    except ImportError:
        pass

    try:
        import greenlet
        from eventlet.patcher import is_monkey_patched  # type: ignore

        greenlet_version = parse_version(greenlet.__version__)

        if greenlet_version is None:
            logger.error(
                "Internal error in Sentry SDK: Could not parse Greenlet version from greenlet.__version__."
            )
            return False

        if is_monkey_patched("thread") and greenlet_version < (0, 5):
            return True
    except ImportError:
        pass

    return False


def _make_threadlocal_contextvars(local: type) -> type:
    class ContextVar:
        # Super-limited impl of ContextVar

        def __init__(self, name: str, default: "Any" = None) -> None:
            self._name = name
            self._default = default
            self._local = local()
            self._original_local = local()

        def get(self, default: "Any" = None) -> "Any":
            return getattr(self._local, "value", default or self._default)

        def set(self, value: "Any") -> "Any":
            token = str(random.getrandbits(64))
            original_value = self.get()
            setattr(self._original_local, token, original_value)
            self._local.value = value
            return token

        def reset(self, token: "Any") -> None:
            self._local.value = getattr(self._original_local, token)
            # delete the original value (this way it works in Python 3.6+)
            del self._original_local.__dict__[token]

    return ContextVar


def _get_contextvars() -> "Tuple[bool, type]":
    """
    Figure out the "right" contextvars installation to use. Returns a
    `contextvars.ContextVar`-like class with a limited API.

    See https://docs.sentry.io/platforms/python/contextvars/ for more information.
    """
    if not _is_contextvars_broken():
        # aiocontextvars is a PyPI package that ensures that the contextvars
        # backport (also a PyPI package) works with asyncio under Python 3.6
        #
        # Import it if available.
        if sys.version_info < (3, 7):
            # `aiocontextvars` is absolutely required for functional
            # contextvars on Python 3.6.
            try:
                from aiocontextvars import ContextVar

                return True, ContextVar
            except ImportError:
                pass
        else:
            # On Python 3.7 contextvars are functional.
            try:
                from contextvars import ContextVar

                return True, ContextVar
            except ImportError:
                pass

    # Fall back to basic thread-local usage.

    from threading import local

    return False, _make_threadlocal_contextvars(local)


HAS_REAL_CONTEXTVARS, ContextVar = _get_contextvars()

CONTEXTVARS_ERROR_MESSAGE = """

With asyncio/ASGI applications, the Sentry SDK requires a functional
installation of `contextvars` to avoid leaking scope/context data across
requests.

Please refer to https://docs.sentry.io/platforms/python/contextvars/ for more information.
"""


def qualname_from_function(func: "Callable[..., Any]") -> "Optional[str]":
    """Return the qualified name of func. Works with regular function, lambda, partial and partialmethod."""
    func_qualname: "Optional[str]" = None

    prefix, suffix = "", ""

    if isinstance(func, partial) and hasattr(func.func, "__name__"):
        prefix, suffix = "partial()"
        func = func.func
    else:
        # The _partialmethod attribute of methods wrapped with partialmethod() was renamed to __partialmethod__ in CPython 3.13:
        # https://github.com/python/cpython/pull/16600
        partial_method = getattr(func, "_partialmethod", None) or getattr(
            func, "__partialmethod__", None
        )
        if isinstance(partial_method, partialmethod):
            prefix, suffix = "partialmethod()"
            func = partial_method.func

    if hasattr(func, "__qualname__"):
        func_qualname = func.__qualname__
    elif hasattr(func, "__name__"):
        func_qualname = func.__name__

    if func_qualname is not None:
        if hasattr(func, "__module__") and isinstance(func.__module__, str):
            func_qualname = func.__module__ + "." + func_qualname
        func_qualname = prefix + func_qualname + suffix

    return func_qualname


def transaction_from_function(func: "Callable[..., Any]") -> "Optional[str]":
    return qualname_from_function(func)


disable_capture_event = ContextVar("disable_capture_event")


class ServerlessTimeoutWarning(Exception):  # noqa: N818
    """Raised when a serverless method is about to reach its timeout."""

    pass


class TimeoutThread(threading.Thread):
    """Creates a Thread which runs (sleeps) for a time duration equal to
    waiting_time and raises a custom ServerlessTimeout exception.
    """

    def __init__(
        self,
        waiting_time: float,
        configured_timeout: int,
        isolation_scope: "Optional[sentry_sdk.Scope]" = None,
        current_scope: "Optional[sentry_sdk.Scope]" = None,
    ) -> None:
        threading.Thread.__init__(self)
        self.waiting_time = waiting_time
        self.configured_timeout = configured_timeout

        self.isolation_scope = isolation_scope
        self.current_scope = current_scope

        self._stop_event = threading.Event()

    def stop(self) -> None:
        self._stop_event.set()

    def _capture_exception(self) -> "ExcInfo":
        exc_info = sys.exc_info()

        client = sentry_sdk.get_client()
        event, hint = event_from_exception(
            exc_info,
            client_options=client.options,
            mechanism={"type": "threading", "handled": False},
        )
        sentry_sdk.capture_event(event, hint=hint)

        return exc_info

    def run(self) -> None:
        self._stop_event.wait(self.waiting_time)

        if self._stop_event.is_set():
            return

        integer_configured_timeout = int(self.configured_timeout)

        # Setting up the exact integer value of configured time(in seconds)
        if integer_configured_timeout < self.configured_timeout:
            integer_configured_timeout = integer_configured_timeout + 1

        # Raising Exception after timeout duration is reached
        if self.isolation_scope is not None and self.current_scope is not None:
            with sentry_sdk.scope.use_isolation_scope(self.isolation_scope):
                with sentry_sdk.scope.use_scope(self.current_scope):
                    try:
                        raise ServerlessTimeoutWarning(
                            "WARNING : Function is expected to get timed out. Configured timeout duration = {} seconds.".format(
                                integer_configured_timeout
                            )
                        )
                    except Exception:
                        reraise(*self._capture_exception())

        raise ServerlessTimeoutWarning(
            "WARNING : Function is expected to get timed out. Configured timeout duration = {} seconds.".format(
                integer_configured_timeout
            )
        )


def to_base64(original: str) -> "Optional[str]":
    """
    Convert a string to base64, via UTF-8. Returns None on invalid input.
    """
    base64_string = None

    try:
        utf8_bytes = original.encode("UTF-8")
        base64_bytes = base64.b64encode(utf8_bytes)
        base64_string = base64_bytes.decode("UTF-8")
    except Exception as err:
        logger.warning("Unable to encode {orig} to base64:".format(orig=original), err)

    return base64_string


def from_base64(base64_string: str) -> "Optional[str]":
    """
    Convert a string from base64, via UTF-8. Returns None on invalid input.
    """
    utf8_string = None

    try:
        only_valid_chars = BASE64_ALPHABET.match(base64_string)
        assert only_valid_chars

        base64_bytes = base64_string.encode("UTF-8")
        utf8_bytes = base64.b64decode(base64_bytes)
        utf8_string = utf8_bytes.decode("UTF-8")
    except Exception as err:
        logger.warning(
            "Unable to decode {b64} from base64:".format(b64=base64_string), err
        )

    return utf8_string


Components = namedtuple("Components", ["scheme", "netloc", "path", "query", "fragment"])


def sanitize_url(
    url: str,
    remove_authority: bool = True,
    remove_query_values: bool = True,
    split: bool = False,
) -> "Union[str, Components]":
    """
    Removes the authority and query parameter values from a given URL.
    """
    parsed_url = urlsplit(url)
    query_params = parse_qs(parsed_url.query, keep_blank_values=True)

    # strip username:password (netloc can be usr:pwd@example.com)
    if remove_authority:
        netloc_parts = parsed_url.netloc.split("@")
        if len(netloc_parts) > 1:
            netloc = "%s:%s@%s" % (
                SENSITIVE_DATA_SUBSTITUTE,
                SENSITIVE_DATA_SUBSTITUTE,
                netloc_parts[-1],
            )
        else:
            netloc = parsed_url.netloc
    else:
        netloc = parsed_url.netloc

    # strip values from query string
    if remove_query_values:
        query_string = unquote(
            urlencode({key: SENSITIVE_DATA_SUBSTITUTE for key in query_params})
        )
    else:
        query_string = parsed_url.query

    components = Components(
        scheme=parsed_url.scheme,
        netloc=netloc,
        query=query_string,
        path=parsed_url.path,
        fragment=parsed_url.fragment,
    )

    if split:
        return components
    else:
        return urlunsplit(components)


ParsedUrl = namedtuple("ParsedUrl", ["url", "query", "fragment"])


def parse_url(url: str, sanitize: bool = True) -> "ParsedUrl":
    """
    Splits a URL into a url (including path), query and fragment. If sanitize is True, the query
    parameters will be sanitized to remove sensitive data. The autority (username and password)
    in the URL will always be removed.
    """
    parsed_url = sanitize_url(
        url, remove_authority=True, remove_query_values=sanitize, split=True
    )

    base_url = urlunsplit(
        Components(
            scheme=parsed_url.scheme,  # type: ignore
            netloc=parsed_url.netloc,  # type: ignore
            query="",
            path=parsed_url.path,  # type: ignore
            fragment="",
        )
    )

    return ParsedUrl(
        url=base_url,
        query=parsed_url.query,  # type: ignore
        fragment=parsed_url.fragment,  # type: ignore
    )


def is_valid_sample_rate(rate: "Any", source: str) -> bool:
    """
    Checks the given sample rate to make sure it is valid type and value (a
    boolean or a number between 0 and 1, inclusive).
    """

    # both booleans and NaN are instances of Real, so a) checking for Real
    # checks for the possibility of a boolean also, and b) we have to check
    # separately for NaN and Decimal does not derive from Real so need to check that too
    if not isinstance(rate, (Real, Decimal)) or math.isnan(rate):
        logger.warning(
            "{source} Given sample rate is invalid. Sample rate must be a boolean or a number between 0 and 1. Got {rate} of type {type}.".format(
                source=source, rate=rate, type=type(rate)
            )
        )
        return False

    # in case rate is a boolean, it will get cast to 1 if it's True and 0 if it's False
    rate = float(rate)
    if rate < 0 or rate > 1:
        logger.warning(
            "{source} Given sample rate is invalid. Sample rate must be between 0 and 1. Got {rate}.".format(
                source=source, rate=rate
            )
        )
        return False

    return True


def match_regex_list(
    item: str,
    regex_list: "Optional[List[str]]" = None,
    substring_matching: bool = False,
) -> bool:
    if regex_list is None:
        return False

    for item_matcher in regex_list:
        if not substring_matching and item_matcher[-1] != "$":
            item_matcher += "$"

        matched = re.search(item_matcher, item)
        if matched:
            return True

    return False


def is_sentry_url(client: "sentry_sdk.client.BaseClient", url: str) -> bool:
    """
    Determines whether the given URL matches the Sentry DSN.
    """
    return (
        client is not None
        and client.transport is not None
        and client.transport.parsed_dsn is not None
        and client.transport.parsed_dsn.netloc in url
    )


def _generate_installed_modules() -> "Iterator[Tuple[str, str]]":
    try:
        from importlib import metadata

        yielded = set()
        for dist in metadata.distributions():
            name = dist.metadata.get("Name", None)  # type: ignore[attr-defined]
            # `metadata` values may be `None`, see:
            # https://github.com/python/cpython/issues/91216
            # and
            # https://github.com/python/importlib_metadata/issues/371
            if name is not None:
                normalized_name = _normalize_module_name(name)
                if dist.version is not None and normalized_name not in yielded:
                    yield normalized_name, dist.version
                    yielded.add(normalized_name)

    except ImportError:
        # < py3.8
        try:
            import pkg_resources
        except ImportError:
            return

        for info in pkg_resources.working_set:
            yield _normalize_module_name(info.key), info.version


def _normalize_module_name(name: str) -> str:
    return name.lower()


def _replace_hyphens_dots_and_underscores_with_dashes(name: str) -> str:
    # https://peps.python.org/pep-0503/#normalized-names
    return re.sub(r"[-_.]+", "-", name)


def _get_installed_modules() -> "Dict[str, str]":
    global _installed_modules
    if _installed_modules is None:
        _installed_modules = dict(_generate_installed_modules())
    return _installed_modules


def package_version(package: str) -> "Optional[Tuple[int, ...]]":
    normalized_package = _normalize_module_name(
        _replace_hyphens_dots_and_underscores_with_dashes(package)
    )

    installed_packages = {
        _replace_hyphens_dots_and_underscores_with_dashes(module): v
        for module, v in _get_installed_modules().items()
    }
    version = installed_packages.get(normalized_package)
    if version is None:
        return None

    return parse_version(version)


def reraise(
    tp: "Optional[Type[BaseException]]",
    value: "Optional[BaseException]",
    tb: "Optional[Any]" = None,
) -> "NoReturn":
    assert value is not None
    if value.__traceback__ is not tb:
        raise value.with_traceback(tb)
    raise value


def _no_op(*_a: "Any", **_k: "Any") -> None:
    """No-op function for ensure_integration_enabled."""
    pass


if TYPE_CHECKING:

    @overload
    def ensure_integration_enabled(
        integration: "type[sentry_sdk.integrations.Integration]",
        original_function: "Callable[P, R]",
    ) -> "Callable[[Callable[P, R]], Callable[P, R]]": ...

    @overload
    def ensure_integration_enabled(
        integration: "type[sentry_sdk.integrations.Integration]",
    ) -> "Callable[[Callable[P, None]], Callable[P, None]]": ...


def ensure_integration_enabled(
    integration: "type[sentry_sdk.integrations.Integration]",
    original_function: "Union[Callable[P, R], Callable[P, None]]" = _no_op,
) -> "Callable[[Callable[P, R]], Callable[P, R]]":
    """
    Ensures a given integration is enabled prior to calling a Sentry-patched function.

    The function takes as its parameters the integration that must be enabled and the original
    function that the SDK is patching. The function returns a function that takes the
    decorated (Sentry-patched) function as its parameter, and returns a function that, when
    called, checks whether the given integration is enabled. If the integration is enabled, the
    function calls the decorated, Sentry-patched function. If the integration is not enabled,
    the original function is called.

    The function also takes care of preserving the original function's signature and docstring.

    Example usage:

    ```python
    @ensure_integration_enabled(MyIntegration, my_function)
    def patch_my_function():
        with sentry_sdk.start_transaction(...):
            return my_function()
    ```
    """
    if TYPE_CHECKING:
        # Type hint to ensure the default function has the right typing. The overloads
        # ensure the default _no_op function is only used when R is None.
        original_function = cast(Callable[P, R], original_function)

    def patcher(sentry_patched_function: "Callable[P, R]") -> "Callable[P, R]":
        def runner(*args: "P.args", **kwargs: "P.kwargs") -> "R":
            if sentry_sdk.get_client().get_integration(integration) is None:
                return original_function(*args, **kwargs)

            return sentry_patched_function(*args, **kwargs)

        if original_function is _no_op:
            return wraps(sentry_patched_function)(runner)

        return wraps(original_function)(runner)

    return patcher


if PY37:

    def nanosecond_time() -> int:
        return time.perf_counter_ns()

else:

    def nanosecond_time() -> int:
        return int(time.perf_counter() * 1e9)


def now() -> float:
    return time.perf_counter()


try:
    from gevent import get_hub as get_gevent_hub
    from gevent.monkey import is_module_patched
except ImportError:
    # it's not great that the signatures are different, get_hub can't return None
    # consider adding an if TYPE_CHECKING to change the signature to Optional[Hub]
    def get_gevent_hub() -> "Optional[Hub]":  # type: ignore[misc]
        return None

    def is_module_patched(mod_name: str) -> bool:
        # unable to import from gevent means no modules have been patched
        return False


def is_gevent() -> bool:
    return is_module_patched("threading") or is_module_patched("_thread")


def get_current_thread_meta(
    thread: "Optional[threading.Thread]" = None,
) -> "Tuple[Optional[int], Optional[str]]":
    """
    Try to get the id of the current thread, with various fall backs.
    """

    # if a thread is specified, that takes priority
    if thread is not None:
        try:
            thread_id = thread.ident
            thread_name = thread.name
            if thread_id is not None:
                return thread_id, thread_name
        except AttributeError:
            pass

    # if the app is using gevent, we should look at the gevent hub first
    # as the id there differs from what the threading module reports
    if is_gevent():
        gevent_hub = get_gevent_hub()
        if gevent_hub is not None:
            try:
                # this is undocumented, so wrap it in try except to be safe
                return gevent_hub.thread_ident, None
            except AttributeError:
                pass

    # use the current thread's id if possible
    try:
        thread = threading.current_thread()
        thread_id = thread.ident
        thread_name = thread.name
        if thread_id is not None:
            return thread_id, thread_name
    except AttributeError:
        pass

    # if we can't get the current thread id, fall back to the main thread id
    try:
        thread = threading.main_thread()
        thread_id = thread.ident
        thread_name = thread.name
        if thread_id is not None:
            return thread_id, thread_name
    except AttributeError:
        pass

    # we've tried everything, time to give up
    return None, None


def _register_control_flow_exception(
    exc_type: "Union[type, list[type], tuple[type], set[type]]",
) -> None:
    if isinstance(exc_type, (list, tuple, set)):
        _control_flow_exception_classes.update(exc_type)
    else:
        _control_flow_exception_classes.add(exc_type)


def should_be_treated_as_error(ty: "Any", value: "Any") -> bool:
    if ty == SystemExit and hasattr(value, "code") and value.code in (0, None):
        # https://docs.python.org/3/library/exceptions.html#SystemExit
        return False

    if issubclass(ty, tuple(_control_flow_exception_classes)):
        return False

    return True


if TYPE_CHECKING:
    T = TypeVar("T")


def try_convert(convert_func: "Callable[[Any], T]", value: "Any") -> "Optional[T]":
    """
    Attempt to convert from an unknown type to a specific type, using the
    given function. Return None if the conversion fails, i.e. if the function
    raises an exception.
    """
    try:
        if isinstance(value, convert_func):  # type: ignore
            return value
    except TypeError:
        pass

    try:
        return convert_func(value)
    except Exception:
        return None


def safe_serialize(data: "Any") -> str:
    """Safely serialize to a readable string."""

    def serialize_item(
        item: "Any",
    ) -> "Union[str, dict[Any, Any], list[Any], tuple[Any, ...]]":
        if callable(item):
            try:
                module = getattr(item, "__module__", None)
                qualname = getattr(item, "__qualname__", None)
                name = getattr(item, "__name__", "anonymous")

                if module and qualname:
                    full_path = f"{module}.{qualname}"
                elif module and name:
                    full_path = f"{module}.{name}"
                else:
                    full_path = name

                return f""
            except Exception:
                return f""
        elif isinstance(item, dict):
            return {k: serialize_item(v) for k, v in item.items()}
        elif isinstance(item, (list, tuple)):
            return [serialize_item(x) for x in item]
        elif hasattr(item, "__dict__"):
            try:
                attrs = {
                    k: serialize_item(v)
                    for k, v in vars(item).items()
                    if not k.startswith("_")
                }
                return f"<{type(item).__name__} {attrs}>"
            except Exception:
                return repr(item)
        else:
            return item

    try:
        serialized = serialize_item(data)
        return (
            json.dumps(serialized, default=str)
            if not isinstance(serialized, str)
            else serialized
        )
    except Exception:
        return str(data)


def has_logs_enabled(options: "Optional[dict[str, Any]]") -> bool:
    if options is None:
        return False

    return bool(
        options.get("enable_logs", False)
        or options["_experiments"].get("enable_logs", False)
    )


def get_before_send_log(
    options: "Optional[dict[str, Any]]",
) -> "Optional[Callable[[Log, Hint], Optional[Log]]]":
    if options is None:
        return None

    return options.get("before_send_log") or options["_experiments"].get(
        "before_send_log"
    )


def has_metrics_enabled(options: "Optional[dict[str, Any]]") -> bool:
    if options is None:
        return False

    return bool(options.get("enable_metrics", True))


def get_before_send_metric(
    options: "Optional[dict[str, Any]]",
) -> "Optional[Callable[[Metric, Hint], Optional[Metric]]]":
    if options is None:
        return None

    return options.get("before_send_metric") or options["_experiments"].get(
        "before_send_metric"
    )


def get_before_send_span(
    options: "Optional[dict[str, Any]]",
) -> "Optional[Callable[[SpanJSON, Hint], Optional[SpanJSON]]]":
    if options is None:
        return None

    return options["_experiments"].get("before_send_span")


def format_attribute(val: "Any") -> "AttributeValue":
    """
    Turn unsupported attribute value types into an AttributeValue.

    We do this as soon as a user-provided attribute is set, to prevent spans,
    logs, metrics and similar from having live references to various objects.

    Note: This is not the final attribute value format. Before they're sent,
    they're serialized further into the actual format the protocol expects:
    https://develop.sentry.dev/sdk/telemetry/attributes/
    """
    if isinstance(val, (bool, int, float, str)):
        return val

    if isinstance(val, (list, tuple)) and not val:
        return []
    elif isinstance(val, list):
        ty = type(val[0])
        if ty in (str, int, float, bool) and all(type(v) is ty for v in val):
            return copy.deepcopy(val)
    elif isinstance(val, tuple):
        ty = type(val[0])
        if ty in (str, int, float, bool) and all(type(v) is ty for v in val):
            return list(val)

    return safe_repr(val)


def serialize_attribute(val: "AttributeValue") -> "SerializedAttributeValue":
    """Serialize attribute value to the transport format."""
    if isinstance(val, bool):
        return {"value": val, "type": "boolean"}
    if isinstance(val, int):
        return {"value": val, "type": "integer"}
    if isinstance(val, float):
        return {"value": val, "type": "double"}
    if isinstance(val, str):
        return {"value": val, "type": "string"}

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

        # Only lists of elements of a single type are supported
        ty = type(val[0])
        if ty in (int, str, bool, float) and all(type(v) is ty for v in val):
            return {"value": val, "type": "array"}

    # Coerce to string if we don't know what to do with the value. This should
    # never happen as we pre-format early in format_attribute, but let's be safe.
    return {"value": safe_repr(val), "type": "string"}