from collections.abc import Mapping
from datetime import datetime, timezone
from importlib import import_module
-
-from sentry_sdk._compat import PY37, check_uwsgi_thread_support
-from sentry_sdk.utils import (
- capture_internal_exceptions,
- current_stacktrace,
- disable_capture_event,
+from typing import cast
+
+from sentry_sdk._compat import PY37, check_uwsgi_thread_support
+from sentry_sdk.utils import (
+ capture_internal_exceptions,
+ current_stacktrace,
format_timestamp,
get_sdk_name,
get_type_name,
@@ -160,7 +160,7 @@ Source code for sentry_sdk.client
from typing import Type
from typing import Union
- from sentry_sdk._types import Event, Hint
+ from sentry_sdk._types import Event, Hint, SDKInfo
from sentry_sdk.integrations import Integration
from sentry_sdk.metrics import MetricsAggregator
from sentry_sdk.scope import Scope
@@ -175,7 +175,7 @@ Source code for sentry_sdk.client
"name": "sentry.python", # SDK name will be overridden after integrations have been loaded with sentry_sdk.integrations.setup_integrations()
"version": VERSION,
"packages": [{"name": "pypi:sentry-sdk", "version": VERSION}],
-}
+} # type: SDKInfo
def _get_options(*args, **kwargs):
@@ -508,532 +508,534 @@ Source code for sentry_sdk.client
try:
setup_continuous_profiler(
self.options,
- capture_func=_capture_envelope,
- )
- except Exception as e:
- logger.debug("Can not set up continuous profiler. (%s)", e)
-
- finally:
- _client_init_debug.set(old_debug)
-
- self._setup_instrumentation(self.options.get("functions_to_trace", []))
-
- if (
- self.monitor
- or self.metrics_aggregator
- or has_profiling_enabled(self.options)
- or isinstance(self.transport, HttpTransport)
- ):
- # If we have anything on that could spawn a background thread, we
- # need to check if it's safe to use them.
- check_uwsgi_thread_support()
-
+ sdk_info=SDK_INFO,
+ capture_func=_capture_envelope,
+ )
+ except Exception as e:
+ logger.debug("Can not set up continuous profiler. (%s)", e)
+
+ finally:
+ _client_init_debug.set(old_debug)
+
+ self._setup_instrumentation(self.options.get("functions_to_trace", []))
+
+ if (
+ self.monitor
+ or self.metrics_aggregator
+ or has_profiling_enabled(self.options)
+ or isinstance(self.transport, HttpTransport)
+ ):
+ # If we have anything on that could spawn a background thread, we
+ # need to check if it's safe to use them.
+ check_uwsgi_thread_support()
+
[docs]
-
def is_active(self):
- # type: () -> bool
- """
- .. versionadded:: 2.0.0
-
- Returns whether the client is active (able to send data to Sentry)
- """
- return True
+ def is_active(self):
+ # type: () -> bool
+ """
+ .. versionadded:: 2.0.0
+
+ Returns whether the client is active (able to send data to Sentry)
+ """
+ return True
[docs]
-
def should_send_default_pii(self):
- # type: () -> bool
- """
- .. versionadded:: 2.0.0
-
- Returns whether the client should send default PII (Personally Identifiable Information) data to Sentry.
- """
- return self.options.get("send_default_pii", False)
+ def should_send_default_pii(self):
+ # type: () -> bool
+ """
+ .. versionadded:: 2.0.0
+
+ Returns whether the client should send default PII (Personally Identifiable Information) data to Sentry.
+ """
+ return self.options.get("send_default_pii", False)
[docs]
-
def capture_event(
- self,
- event, # type: Event
- hint=None, # type: Optional[Hint]
- scope=None, # type: Optional[Scope]
- ):
- # type: (...) -> Optional[str]
- """Captures an event.
-
- :param event: A ready-made event that can be directly sent to Sentry.
-
- :param hint: Contains metadata about the event that can be read from `before_send`, such as the original exception object or a HTTP request object.
-
- :param scope: An optional :py:class:`sentry_sdk.Scope` to apply to events.
-
- :returns: An event ID. May be `None` if there is no DSN set or of if the SDK decided to discard the event for other reasons. In such situations setting `debug=True` on `init()` may help.
- """
- if disable_capture_event.get(False):
- return None
+ def capture_event(
+ self,
+ event, # type: Event
+ hint=None, # type: Optional[Hint]
+ scope=None, # type: Optional[Scope]
+ ):
+ # type: (...) -> Optional[str]
+ """Captures an event.
+
+ :param event: A ready-made event that can be directly sent to Sentry.
+
+ :param hint: Contains metadata about the event that can be read from `before_send`, such as the original exception object or a HTTP request object.
+
+ :param scope: An optional :py:class:`sentry_sdk.Scope` to apply to events.
- if hint is None:
- hint = {}
- event_id = event.get("event_id")
- hint = dict(hint or ()) # type: Hint
-
- if event_id is None:
- event["event_id"] = event_id = uuid.uuid4().hex
- if not self._should_capture(event, hint, scope):
- return None
-
- profile = event.pop("profile", None)
+ :returns: An event ID. May be `None` if there is no DSN set or of if the SDK decided to discard the event for other reasons. In such situations setting `debug=True` on `init()` may help.
+ """
+ if hint is None:
+ hint = {}
+ event_id = event.get("event_id")
+ hint = dict(hint or ()) # type: Hint
+
+ if event_id is None:
+ event["event_id"] = event_id = uuid.uuid4().hex
+ if not self._should_capture(event, hint, scope):
+ return None
- event_opt = self._prepare_event(event, hint, scope)
- if event_opt is None:
- return None
-
- # whenever we capture an event we also check if the session needs
- # to be updated based on that information.
- session = scope._session if scope else None
- if session:
- self._update_session_from_event(session, event)
-
- is_transaction = event_opt.get("type") == "transaction"
- is_checkin = event_opt.get("type") == "check_in"
-
- if (
- not is_transaction
- and not is_checkin
- and not self._should_sample_error(event, hint)
- ):
- return None
-
- attachments = hint.get("attachments")
+ profile = event.pop("profile", None)
+
+ event_opt = self._prepare_event(event, hint, scope)
+ if event_opt is None:
+ return None
+
+ # whenever we capture an event we also check if the session needs
+ # to be updated based on that information.
+ session = scope._session if scope else None
+ if session:
+ self._update_session_from_event(session, event)
+
+ is_transaction = event_opt.get("type") == "transaction"
+ is_checkin = event_opt.get("type") == "check_in"
+
+ if (
+ not is_transaction
+ and not is_checkin
+ and not self._should_sample_error(event, hint)
+ ):
+ return None
- trace_context = event_opt.get("contexts", {}).get("trace") or {}
- dynamic_sampling_context = trace_context.pop("dynamic_sampling_context", {})
-
- headers = {
- "event_id": event_opt["event_id"],
- "sent_at": format_timestamp(datetime.now(timezone.utc)),
- } # type: dict[str, object]
-
- if dynamic_sampling_context:
- headers["trace"] = dynamic_sampling_context
-
- envelope = Envelope(headers=headers)
+ attachments = hint.get("attachments")
+
+ trace_context = event_opt.get("contexts", {}).get("trace") or {}
+ dynamic_sampling_context = trace_context.pop("dynamic_sampling_context", {})
+
+ headers = {
+ "event_id": event_opt["event_id"],
+ "sent_at": format_timestamp(datetime.now(timezone.utc)),
+ } # type: dict[str, object]
+
+ if dynamic_sampling_context:
+ headers["trace"] = dynamic_sampling_context
- if is_transaction:
- if isinstance(profile, Profile):
- envelope.add_profile(profile.to_json(event_opt, self.options))
- envelope.add_transaction(event_opt)
- elif is_checkin:
- envelope.add_checkin(event_opt)
- else:
- envelope.add_event(event_opt)
-
- for attachment in attachments or ():
- envelope.add_item(attachment.to_envelope_item())
-
- if self.spotlight:
- self.spotlight.capture_envelope(envelope)
-
- if self.transport is None:
- return None
-
- self.transport.capture_envelope(envelope)
+ envelope = Envelope(headers=headers)
+
+ if is_transaction:
+ if isinstance(profile, Profile):
+ envelope.add_profile(profile.to_json(event_opt, self.options))
+ envelope.add_transaction(event_opt)
+ elif is_checkin:
+ envelope.add_checkin(event_opt)
+ else:
+ envelope.add_event(event_opt)
+
+ for attachment in attachments or ():
+ envelope.add_item(attachment.to_envelope_item())
+
+ if self.spotlight:
+ self.spotlight.capture_envelope(envelope)
+
+ if self.transport is None:
+ return None
- return event_id
-
+ self.transport.capture_envelope(envelope)
- def capture_session(
- self, session # type: Session
- ):
- # type: (...) -> None
- if not session.release:
- logger.info("Discarded session update because of missing release")
- else:
- self.session_flusher.add_session(session)
-
+ return event_id
[docs]
-
def get_integration(
- self, name_or_class # type: Union[str, Type[Integration]]
- ):
- # type: (...) -> Any
- """Returns the integration for this client by name or class.
- If the client does not have that integration then `None` is returned.
- """
- if isinstance(name_or_class, str):
- integration_name = name_or_class
- elif name_or_class.identifier is not None:
- integration_name = name_or_class.identifier
- else:
- raise ValueError("Integration has no name")
-
- return self.integrations.get(integration_name)
-
+ def get_integration(
+ self, name_or_class # type: Union[str, Type[Integration]]
+ ):
+ # type: (...) -> Any
+ """Returns the integration for this client by name or class.
+ If the client does not have that integration then `None` is returned.
+ """
+ if isinstance(name_or_class, str):
+ integration_name = name_or_class
+ elif name_or_class.identifier is not None:
+ integration_name = name_or_class.identifier
+ else:
+ raise ValueError("Integration has no name")
+ return self.integrations.get(integration_name)