sentry_sdk.hub

import warnings
from contextlib import contextmanager
from typing import TYPE_CHECKING

from sentry_sdk import (
    get_client,
    get_current_scope,
    get_global_scope,
    get_isolation_scope,
)
from sentry_sdk._compat import with_metaclass
from sentry_sdk.client import Client
from sentry_sdk.consts import INSTRUMENTER
from sentry_sdk.scope import _ScopeManager
from sentry_sdk.tracing import (
    NoOpSpan,
    Span,
    Transaction,
)
from sentry_sdk.utils import (
    ContextVar,
    logger,
)

if TYPE_CHECKING:
    from typing import (
        Any,
        Callable,
        ContextManager,
        Dict,
        Generator,
        List,
        Optional,
        Tuple,
        Type,
        TypeVar,
        Union,
        overload,
    )

    from typing_extensions import Unpack

    from sentry_sdk._types import (
        Breadcrumb,
        BreadcrumbHint,
        Event,
        ExcInfo,
        Hint,
        LogLevelStr,
        SamplingContext,
    )
    from sentry_sdk.client import BaseClient
    from sentry_sdk.integrations import Integration
    from sentry_sdk.scope import Scope
    from sentry_sdk.tracing import TransactionKwargs

    T = TypeVar("T")

else:

    def overload(x: "T") -> "T":
        return x


class SentryHubDeprecationWarning(DeprecationWarning):
    """
    A custom deprecation warning to inform users that the Hub is deprecated.
    """

    _MESSAGE = (
        "`sentry_sdk.Hub` is deprecated and will be removed in a future major release. "
        "Please consult our 1.x to 2.x migration guide for details on how to migrate "
        "`Hub` usage to the new API: "
        "https://docs.sentry.io/platforms/python/migration/1.x-to-2.x"
    )

    def __init__(self, *_: object) -> None:
        super().__init__(self._MESSAGE)


@contextmanager
def _suppress_hub_deprecation_warning() -> "Generator[None, None, None]":
    """Utility function to suppress deprecation warnings for the Hub."""
    with warnings.catch_warnings():
        warnings.filterwarnings("ignore", category=SentryHubDeprecationWarning)
        yield


_local = ContextVar("sentry_current_hub")


class HubMeta(type):
    @property
    def current(cls) -> "Hub":
        """Returns the current instance of the hub."""
        warnings.warn(SentryHubDeprecationWarning(), stacklevel=2)
        rv = _local.get(None)
        if rv is None:
            with _suppress_hub_deprecation_warning():
                # This will raise a deprecation warning; suppress it since we already warned above.
                rv = Hub(GLOBAL_HUB)
            _local.set(rv)
        return rv

    @property
    def main(cls) -> "Hub":
        """Returns the main instance of the hub."""
        warnings.warn(SentryHubDeprecationWarning(), stacklevel=2)
        return GLOBAL_HUB


class Hub(with_metaclass(HubMeta)):  # type: ignore
    """
    .. deprecated:: 2.0.0
        The Hub is deprecated. Its functionality will be merged into :py:class:`sentry_sdk.scope.Scope`.

    The hub wraps the concurrency management of the SDK.  Each thread has
    its own hub but the hub might transfer with the flow of execution if
    context vars are available.

    If the hub is used with a with statement it's temporarily activated.
    """

    _stack: "List[Tuple[Optional[Client], Scope]]" = None  # type: ignore[assignment]
    _scope: "Optional[Scope]" = None

    # Mypy doesn't pick up on the metaclass.

    if TYPE_CHECKING:
        current: "Hub" = None  # type: ignore[assignment]
        main: "Optional[Hub]" = None

    def __init__(
        self,
        client_or_hub: "Optional[Union[Hub, Client]]" = None,
        scope: "Optional[Any]" = None,
    ) -> None:
        warnings.warn(SentryHubDeprecationWarning(), stacklevel=2)

        current_scope = None

        if isinstance(client_or_hub, Hub):
            client = get_client()
            if scope is None:
                # hub cloning is going on, we use a fork of the current/isolation scope for context manager
                scope = get_isolation_scope().fork()
                current_scope = get_current_scope().fork()
        else:
            client = client_or_hub  # type: ignore
            get_global_scope().set_client(client)

        if scope is None:  # so there is no Hub cloning going on
            # just the current isolation scope is used for context manager
            scope = get_isolation_scope()
            current_scope = get_current_scope()

        if current_scope is None:
            # just the current current scope is used for context manager
            current_scope = get_current_scope()

        self._stack = [(client, scope)]  # type: ignore
        self._last_event_id: "Optional[str]" = None
        self._old_hubs: "List[Hub]" = []

        self._old_current_scopes: "List[Scope]" = []
        self._old_isolation_scopes: "List[Scope]" = []
        self._current_scope: "Scope" = current_scope
        self._scope: "Scope" = scope

    def __enter__(self) -> "Hub":
        self._old_hubs.append(Hub.current)
        _local.set(self)

        current_scope = get_current_scope()
        self._old_current_scopes.append(current_scope)
        scope._current_scope.set(self._current_scope)

        isolation_scope = get_isolation_scope()
        self._old_isolation_scopes.append(isolation_scope)
        scope._isolation_scope.set(self._scope)

        return self

    def __exit__(
        self,
        exc_type: "Optional[type]",
        exc_value: "Optional[BaseException]",
        tb: "Optional[Any]",
    ) -> None:
        old = self._old_hubs.pop()
        _local.set(old)

        old_current_scope = self._old_current_scopes.pop()
        scope._current_scope.set(old_current_scope)

        old_isolation_scope = self._old_isolation_scopes.pop()
        scope._isolation_scope.set(old_isolation_scope)

    def run(
        self,
        callback: "Callable[[], T]",
    ) -> "T":
        """
        .. deprecated:: 2.0.0
            This function is deprecated and will be removed in a future release.

        Runs a callback in the context of the hub.  Alternatively the
        with statement can be used on the hub directly.
        """
        with self:
            return callback()