# -*- coding: utf-8 -*-
"""
Sentry SDK core primitives.
This module provides the configuration dataclass, SDK initialization, and
the main event-lifecycle functions used directly or by the
decorator layer.
Classes:
SentryConfig: Dataclass holding all parameters for ``sentry_sdk.init``.
Functions:
init_sentry: Initialize the Sentry SDK from a ``SentryConfig``.
flush_sentry: Block until pending events are delivered (or timeout).
capture_exception: Capture an exception with optional per-event tags/extra.
capture_message: Capture a message with optional severity, tags, and extra.
Example:
Basic initialization and manual event capture:
.. code-block:: python
from core_sentry.base import SentryConfig, init_sentry, capture_exception, flush_sentry
init_sentry(SentryConfig(
dsn="https://...",
env="production",
tags={"service": "etl-pipeline", "version": "1.0.0"},
))
try:
run_job()
except Exception as exc:
capture_exception(exc, tags={"job": "daily-sync"}, extra={"rows": 9999})
finally:
flush_sentry(timeout=5.0)
..
"""
from dataclasses import dataclass
from dataclasses import field
from typing import Any
from typing import Dict
from typing import Literal
from typing import Optional
import sentry_sdk
[docs]
@dataclass
class SentryConfig:
"""
Configuration for Sentry SDK initialization.
:param dsn: The DSN tells the SDK where to send the events.
:param env: Environments tell you where an error occurred (production, staging).
:param tags:
Tags are key/value string pairs that are both indexed and searchable. Tags power
features in sentry.io such as filters and tag-distribution maps. Tags also help you quickly
both access related events and view the tag distribution for a set of events.
More Info: https://docs.sentry.io/platforms/python/enriching-events/tags/
:param traces_sample_rate:
Controls the percentage of transactions to trace (0.0 to 1.0).
Defaults to 0.1 (10%) for better performance in production.
:param options:
Additional keyword arguments forwarded to ``sentry_sdk.init``.
See: https://docs.sentry.io/platforms/python/configuration/options/
Example::
config = SentryConfig(
dsn=os.environ["SENTRY_DSN"],
env="production",
tags={"service": "etl-pipeline", "version": "1.0.0"},
traces_sample_rate=0.5,
)
init_sentry(config)
"""
dsn: str
env: str
tags: Optional[Dict[str, str]] = None
traces_sample_rate: float = 0.1
options: Dict[str, Any] = field(default_factory=dict)
[docs]
def init_sentry(config: SentryConfig) -> None:
"""
Initialize the Sentry SDK from a ``SentryConfig``. Skips initialization
if the SDK is already configured. Tags are always applied on every call,
even when initialization is skipped.
:param config: Sentry configuration object.
Example::
init_sentry(SentryConfig(
dsn=os.environ["SENTRY_DSN"],
env="production",
tags={"service": "etl-pipeline", "version": "1.0.0"},
traces_sample_rate=0.5,
))
"""
if not sentry_sdk.get_client().dsn:
sentry_sdk.init(
dsn=config.dsn,
environment=config.env,
traces_sample_rate=config.traces_sample_rate,
**config.options,
)
if config.tags:
for tag, value in config.tags.items():
sentry_sdk.set_tag(key=tag, value=value)
[docs]
def flush_sentry(timeout: float = 2.0) -> bool:
"""
Flush pending Sentry events. I should be called before
process exit in scripts/ETLs.
:param timeout: Maximum seconds to wait for the queue to drain.
:returns: ``False`` if the timeout expires before all events are sent.
Example::
flushed = flush_sentry(timeout=5.0)
if not flushed:
print("WARNING: some Sentry events may not have been delivered.")
"""
flushed = bool(sentry_sdk.flush(timeout=timeout))
return flushed
[docs]
def capture_exception(
error: BaseException,
tags: Optional[Dict[str, str]] = None,
extra: Optional[Dict[str, Any]] = None,
) -> Optional[str]:
"""
Capture an exception and send it to Sentry.
:param error: Exception to capture.
:param tags: Per-event tags (do not affect subsequent captures).
:param extra: Per-event extra context (do not affect subsequent captures).
:returns: Sentry event ID, or ``None`` if the SDK is not initialized.
Example::
try:
raise ValueError("Unexpected null value in column 'amount'")
except ValueError as exc:
event_id = capture_exception(
exc,
tags={"job": "daily-sync", "stage": "transform"},
extra={"row_index": 1042, "column": "amount"},
)
"""
with sentry_sdk.new_scope() as scope:
_set_tags_to_scope(scope, tags)
_set_extras_to_scope(scope, extra)
return sentry_sdk.capture_exception(error)
[docs]
def capture_message(
message: str,
level: Literal["fatal", "critical", "error", "warning", "info", "debug"] = "info",
tags: Optional[Dict[str, str]] = None,
extra: Optional[Dict[str, Any]] = None,
) -> Optional[str]:
"""
Capture a message and send it to Sentry.
:param message: Message text to send.
:param level: Sentry severity level ("debug", "info", "warning", "error", "fatal").
:param tags: Per-event tags (do not affect subsequent captures).
:param extra: Per-event extra context (do not affect subsequent captures).
:returns: Sentry event ID, or ``None`` if the SDK is not initialized.
Example::
event_id = capture_message(
"ETL job started",
level="info",
tags={"job": "daily-sync"},
extra={"source": "postgres", "destination": "bigquery"},
)
"""
with sentry_sdk.new_scope() as scope:
_set_tags_to_scope(scope, tags)
_set_extras_to_scope(scope, extra)
return sentry_sdk.capture_message(message, level=level)
def _set_tags_to_scope(
scope: sentry_sdk.Scope,
tags: Optional[Dict[str, str]] = None,
) -> None:
if tags:
for key, value in tags.items():
scope.set_tag(key, value)
def _set_extras_to_scope(
scope: sentry_sdk.Scope,
extra: Optional[Dict[str, Any]] = None,
) -> None:
if extra:
for key, value in extra.items():
scope.set_extra(key, value)