| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555 |
- import inspect
- import warnings
- from contextlib import contextmanager
- from sentry_sdk import tracing_utils, Client
- from sentry_sdk._init_implementation import init
- from sentry_sdk.consts import INSTRUMENTER
- from sentry_sdk.scope import Scope, _ScopeManager, new_scope, isolation_scope
- from sentry_sdk.tracing import NoOpSpan, Transaction, trace
- from sentry_sdk.crons import monitor
- from typing import TYPE_CHECKING
- if TYPE_CHECKING:
- from collections.abc import Mapping
- from typing import Any
- from typing import Dict
- from typing import Generator
- from typing import Optional
- from typing import overload
- from typing import Callable
- from typing import TypeVar
- from typing import ContextManager
- from typing import Union
- from typing_extensions import Unpack
- from sentry_sdk.client import BaseClient
- from sentry_sdk._types import (
- Event,
- Hint,
- Breadcrumb,
- BreadcrumbHint,
- ExcInfo,
- MeasurementUnit,
- LogLevelStr,
- SamplingContext,
- )
- from sentry_sdk.tracing import Span, TransactionKwargs
- T = TypeVar("T")
- F = TypeVar("F", bound=Callable[..., Any])
- else:
- def overload(x):
- # type: (T) -> T
- return x
- # When changing this, update __all__ in __init__.py too
- __all__ = [
- "init",
- "add_attachment",
- "add_breadcrumb",
- "capture_event",
- "capture_exception",
- "capture_message",
- "configure_scope",
- "continue_trace",
- "flush",
- "get_baggage",
- "get_client",
- "get_global_scope",
- "get_isolation_scope",
- "get_current_scope",
- "get_current_span",
- "get_traceparent",
- "is_initialized",
- "isolation_scope",
- "last_event_id",
- "new_scope",
- "push_scope",
- "set_context",
- "set_extra",
- "set_level",
- "set_measurement",
- "set_tag",
- "set_tags",
- "set_user",
- "start_span",
- "start_transaction",
- "trace",
- "monitor",
- "start_session",
- "end_session",
- "set_transaction_name",
- "update_current_span",
- ]
- def scopemethod(f):
- # type: (F) -> F
- f.__doc__ = "%s\n\n%s" % (
- "Alias for :py:meth:`sentry_sdk.Scope.%s`" % f.__name__,
- inspect.getdoc(getattr(Scope, f.__name__)),
- )
- return f
- def clientmethod(f):
- # type: (F) -> F
- f.__doc__ = "%s\n\n%s" % (
- "Alias for :py:meth:`sentry_sdk.Client.%s`" % f.__name__,
- inspect.getdoc(getattr(Client, f.__name__)),
- )
- return f
- @scopemethod
- def get_client():
- # type: () -> BaseClient
- return Scope.get_client()
- def is_initialized():
- # type: () -> bool
- """
- .. versionadded:: 2.0.0
- Returns whether Sentry has been initialized or not.
- If a client is available and the client is active
- (meaning it is configured to send data) then
- Sentry is initialized.
- """
- return get_client().is_active()
- @scopemethod
- def get_global_scope():
- # type: () -> Scope
- return Scope.get_global_scope()
- @scopemethod
- def get_isolation_scope():
- # type: () -> Scope
- return Scope.get_isolation_scope()
- @scopemethod
- def get_current_scope():
- # type: () -> Scope
- return Scope.get_current_scope()
- @scopemethod
- def last_event_id():
- # type: () -> Optional[str]
- """
- See :py:meth:`sentry_sdk.Scope.last_event_id` documentation regarding
- this method's limitations.
- """
- return Scope.last_event_id()
- @scopemethod
- def capture_event(
- event, # type: Event
- hint=None, # type: Optional[Hint]
- scope=None, # type: Optional[Any]
- **scope_kwargs, # type: Any
- ):
- # type: (...) -> Optional[str]
- return get_current_scope().capture_event(event, hint, scope=scope, **scope_kwargs)
- @scopemethod
- def capture_message(
- message, # type: str
- level=None, # type: Optional[LogLevelStr]
- scope=None, # type: Optional[Any]
- **scope_kwargs, # type: Any
- ):
- # type: (...) -> Optional[str]
- return get_current_scope().capture_message(
- message, level, scope=scope, **scope_kwargs
- )
- @scopemethod
- def capture_exception(
- error=None, # type: Optional[Union[BaseException, ExcInfo]]
- scope=None, # type: Optional[Any]
- **scope_kwargs, # type: Any
- ):
- # type: (...) -> Optional[str]
- return get_current_scope().capture_exception(error, scope=scope, **scope_kwargs)
- @scopemethod
- def add_attachment(
- bytes=None, # type: Union[None, bytes, Callable[[], bytes]]
- filename=None, # type: Optional[str]
- path=None, # type: Optional[str]
- content_type=None, # type: Optional[str]
- add_to_transactions=False, # type: bool
- ):
- # type: (...) -> None
- return get_isolation_scope().add_attachment(
- bytes, filename, path, content_type, add_to_transactions
- )
- @scopemethod
- def add_breadcrumb(
- crumb=None, # type: Optional[Breadcrumb]
- hint=None, # type: Optional[BreadcrumbHint]
- **kwargs, # type: Any
- ):
- # type: (...) -> None
- return get_isolation_scope().add_breadcrumb(crumb, hint, **kwargs)
- @overload
- def configure_scope():
- # type: () -> ContextManager[Scope]
- pass
- @overload
- def configure_scope( # noqa: F811
- callback, # type: Callable[[Scope], None]
- ):
- # type: (...) -> None
- pass
- def configure_scope( # noqa: F811
- callback=None, # type: Optional[Callable[[Scope], None]]
- ):
- # type: (...) -> Optional[ContextManager[Scope]]
- """
- Reconfigures the scope.
- :param callback: If provided, call the callback with the current scope.
- :returns: If no callback is provided, returns a context manager that returns the scope.
- """
- warnings.warn(
- "sentry_sdk.configure_scope is deprecated and will be removed in the next major version. "
- "Please consult our migration guide to learn how to migrate to the new API: "
- "https://docs.sentry.io/platforms/python/migration/1.x-to-2.x#scope-configuring",
- DeprecationWarning,
- stacklevel=2,
- )
- scope = get_isolation_scope()
- scope.generate_propagation_context()
- if callback is not None:
- # TODO: used to return None when client is None. Check if this changes behavior.
- callback(scope)
- return None
- @contextmanager
- def inner():
- # type: () -> Generator[Scope, None, None]
- yield scope
- return inner()
- @overload
- def push_scope():
- # type: () -> ContextManager[Scope]
- pass
- @overload
- def push_scope( # noqa: F811
- callback, # type: Callable[[Scope], None]
- ):
- # type: (...) -> None
- pass
- def push_scope( # noqa: F811
- callback=None, # type: Optional[Callable[[Scope], None]]
- ):
- # type: (...) -> Optional[ContextManager[Scope]]
- """
- Pushes a new layer on the scope stack.
- :param callback: If provided, this method pushes a scope, calls
- `callback`, and pops the scope again.
- :returns: If no `callback` is provided, a context manager that should
- be used to pop the scope again.
- """
- warnings.warn(
- "sentry_sdk.push_scope is deprecated and will be removed in the next major version. "
- "Please consult our migration guide to learn how to migrate to the new API: "
- "https://docs.sentry.io/platforms/python/migration/1.x-to-2.x#scope-pushing",
- DeprecationWarning,
- stacklevel=2,
- )
- if callback is not None:
- with warnings.catch_warnings():
- warnings.simplefilter("ignore", DeprecationWarning)
- with push_scope() as scope:
- callback(scope)
- return None
- return _ScopeManager()
- @scopemethod
- def set_tag(key, value):
- # type: (str, Any) -> None
- return get_isolation_scope().set_tag(key, value)
- @scopemethod
- def set_tags(tags):
- # type: (Mapping[str, object]) -> None
- return get_isolation_scope().set_tags(tags)
- @scopemethod
- def set_context(key, value):
- # type: (str, Dict[str, Any]) -> None
- return get_isolation_scope().set_context(key, value)
- @scopemethod
- def set_extra(key, value):
- # type: (str, Any) -> None
- return get_isolation_scope().set_extra(key, value)
- @scopemethod
- def set_user(value):
- # type: (Optional[Dict[str, Any]]) -> None
- return get_isolation_scope().set_user(value)
- @scopemethod
- def set_level(value):
- # type: (LogLevelStr) -> None
- return get_isolation_scope().set_level(value)
- @clientmethod
- def flush(
- timeout=None, # type: Optional[float]
- callback=None, # type: Optional[Callable[[int, float], None]]
- ):
- # type: (...) -> None
- return get_client().flush(timeout=timeout, callback=callback)
- @scopemethod
- def start_span(
- **kwargs, # type: Any
- ):
- # type: (...) -> Span
- return get_current_scope().start_span(**kwargs)
- @scopemethod
- def start_transaction(
- transaction=None, # type: Optional[Transaction]
- instrumenter=INSTRUMENTER.SENTRY, # type: str
- custom_sampling_context=None, # type: Optional[SamplingContext]
- **kwargs, # type: Unpack[TransactionKwargs]
- ):
- # type: (...) -> Union[Transaction, NoOpSpan]
- """
- Start and return a transaction on the current scope.
- Start an existing transaction if given, otherwise create and start a new
- transaction with kwargs.
- This is the entry point to manual tracing instrumentation.
- A tree structure can be built by adding child spans to the transaction,
- and child spans to other spans. To start a new child span within the
- transaction or any span, call the respective `.start_child()` method.
- Every child span must be finished before the transaction is finished,
- otherwise the unfinished spans are discarded.
- When used as context managers, spans and transactions are automatically
- finished at the end of the `with` block. If not using context managers,
- call the `.finish()` method.
- When the transaction is finished, it will be sent to Sentry with all its
- finished child spans.
- :param transaction: The transaction to start. If omitted, we create and
- start a new transaction.
- :param instrumenter: This parameter is meant for internal use only. It
- will be removed in the next major version.
- :param custom_sampling_context: The transaction's custom sampling context.
- :param kwargs: Optional keyword arguments to be passed to the Transaction
- constructor. See :py:class:`sentry_sdk.tracing.Transaction` for
- available arguments.
- """
- return get_current_scope().start_transaction(
- transaction, instrumenter, custom_sampling_context, **kwargs
- )
- def set_measurement(name, value, unit=""):
- # type: (str, float, MeasurementUnit) -> None
- """
- .. deprecated:: 2.28.0
- This function is deprecated and will be removed in the next major release.
- """
- transaction = get_current_scope().transaction
- if transaction is not None:
- transaction.set_measurement(name, value, unit)
- def get_current_span(scope=None):
- # type: (Optional[Scope]) -> Optional[Span]
- """
- Returns the currently active span if there is one running, otherwise `None`
- """
- return tracing_utils.get_current_span(scope)
- def get_traceparent():
- # type: () -> Optional[str]
- """
- Returns the traceparent either from the active span or from the scope.
- """
- return get_current_scope().get_traceparent()
- def get_baggage():
- # type: () -> Optional[str]
- """
- Returns Baggage either from the active span or from the scope.
- """
- baggage = get_current_scope().get_baggage()
- if baggage is not None:
- return baggage.serialize()
- return None
- def continue_trace(
- environ_or_headers, op=None, name=None, source=None, origin="manual"
- ):
- # type: (Dict[str, Any], Optional[str], Optional[str], Optional[str], str) -> Transaction
- """
- Sets the propagation context from environment or headers and returns a transaction.
- """
- return get_isolation_scope().continue_trace(
- environ_or_headers, op, name, source, origin
- )
- @scopemethod
- def start_session(
- session_mode="application", # type: str
- ):
- # type: (...) -> None
- return get_isolation_scope().start_session(session_mode=session_mode)
- @scopemethod
- def end_session():
- # type: () -> None
- return get_isolation_scope().end_session()
- @scopemethod
- def set_transaction_name(name, source=None):
- # type: (str, Optional[str]) -> None
- return get_current_scope().set_transaction_name(name, source)
- def update_current_span(op=None, name=None, attributes=None, data=None):
- # type: (Optional[str], Optional[str], Optional[dict[str, Union[str, int, float, bool]]], Optional[dict[str, Any]]) -> None
- """
- Update the current active span with the provided parameters.
- This function allows you to modify properties of the currently active span.
- If no span is currently active, this function will do nothing.
- :param op: The operation name for the span. This is a high-level description
- of what the span represents (e.g., "http.client", "db.query").
- You can use predefined constants from :py:class:`sentry_sdk.consts.OP`
- or provide your own string. If not provided, the span's operation will
- remain unchanged.
- :type op: str or None
- :param name: The human-readable name/description for the span. This provides
- more specific details about what the span represents (e.g., "GET /api/users",
- "SELECT * FROM users"). If not provided, the span's name will remain unchanged.
- :type name: str or None
- :param data: A dictionary of key-value pairs to add as data to the span. This
- data will be merged with any existing span data. If not provided,
- no data will be added.
- .. deprecated:: 2.35.0
- Use ``attributes`` instead. The ``data`` parameter will be removed
- in a future version.
- :type data: dict[str, Union[str, int, float, bool]] or None
- :param attributes: A dictionary of key-value pairs to add as attributes to the span.
- Attribute values must be strings, integers, floats, or booleans. These
- attributes will be merged with any existing span data. If not provided,
- no attributes will be added.
- :type attributes: dict[str, Union[str, int, float, bool]] or None
- :returns: None
- .. versionadded:: 2.35.0
- Example::
- import sentry_sdk
- from sentry_sdk.consts import OP
- sentry_sdk.update_current_span(
- op=OP.FUNCTION,
- name="process_user_data",
- attributes={"user_id": 123, "batch_size": 50}
- )
- """
- current_span = get_current_span()
- if current_span is None:
- return
- if op is not None:
- current_span.op = op
- if name is not None:
- # internally it is still description
- current_span.description = name
- if data is not None and attributes is not None:
- raise ValueError(
- "Cannot provide both `data` and `attributes`. Please use only `attributes`."
- )
- if data is not None:
- warnings.warn(
- "The `data` parameter is deprecated. Please use `attributes` instead.",
- DeprecationWarning,
- stacklevel=2,
- )
- attributes = data
- if attributes is not None:
- current_span.update_data(attributes)
|