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()