# -*- coding: utf-8 -*-
"""
Sentry SDK initialization decorators.
This module provides decorators for initializing the Sentry SDK with automatic
error tracking and monitoring capabilities. The decorators handle Sentry
initialization, tag setting, exception capture, and event flushing.
Functions:
with_sentry: Decorator that initializes Sentry SDK and sets tags for a function.
Example:
Basic usage:
.. code-block:: python
from core_sentry.base import SentryConfig
from core_sentry.decorators.base import with_sentry
@with_sentry(config=SentryConfig(dsn="https://...", env="production"))
def my_function():
return "Hello World"
..
ETL/script usage with automatic exception capture and flush on exit:
.. code-block:: python
@with_sentry(
config=SentryConfig(
dsn="https://...",
env="production",
tags={"service": "etl-pipeline", "version": "1.0.0"},
traces_sample_rate=0.5,
),
capture_exceptions=True,
flush=True,
flush_timeout=5.0,
)
def run_etl():
pass
..
Features:
- Automatic Sentry SDK initialization with provided configuration
- Tag management for enhanced error tracking and filtering
- Prevention of duplicate initialization when Sentry is already configured
- Support for all standard Sentry SDK configuration options
- Optional automatic exception capture before re-raising
- Optional event flush on function exit (success or failure)
"""
from __future__ import annotations
import functools
import typing
from dataclasses import dataclass
from core_sentry.base import SentryConfig
from core_sentry.base import capture_exception as _capture_exception
from core_sentry.base import flush_sentry as _flush_sentry
from core_sentry.base import init_sentry as _init_sentry
[docs]
@dataclass
class SentryRuntimeConfig:
"""
Runtime behavior configuration for the ``init_sentry`` decorator.
:param capture_exceptions:
When ``True``, any unhandled exception raised by the
wrapped function is captured and sent to Sentry before being re-raised.
Defaults to ``False``.
:param flush:
When ``True``, pending Sentry events are flushed after the
wrapped function exits (whether it succeeded or raised). Recommended
for short-lived scripts and ETLs. Defaults to ``False``.
:param flush_timeout:
Maximum seconds to wait for the event queue to drain
when ``flush=True``. Defaults to ``2.0``.
:param tags:
Per-capture tags attached to every exception event sent via this
decorator. Do not affect subsequent captures outside the decorator.
:param extra:
Per-capture extra context attached to every exception event
sent via this decorator. Do not affect subsequent captures outside the
decorator.
Example::
SentryRuntimeConfig(
capture_exceptions=True,
flush=True,
flush_timeout=5.0,
tags={"job": "daily-sync"},
extra={"rows_processed": 9999},
)
"""
capture_exceptions: bool = False
flush: bool = False
flush_timeout: float = 2.0
tags: typing.Optional[typing.Dict[str, str]] = None
extra: typing.Optional[typing.Dict[str, typing.Any]] = None
[docs]
def with_sentry(
_func: typing.Optional[typing.Callable] = None,
*,
config: SentryConfig,
runtime_config: SentryRuntimeConfig = SentryRuntimeConfig(),
) -> typing.Callable | typing.Callable[[typing.Callable], typing.Callable]:
"""
Decorator that initializes the Sentry SDK before the wrapped function runs.
Basic usage:
.. code-block:: python
@with_sentry(config=SentryConfig(dsn="SomeDSN", env="production"))
def my_function():
pass
..
Use ``capture_exceptions=True`` and ``flush=True`` on functions where you want
to ensure proper error tracking, every unhandled exception is explicitly
captured with its full context and all events are guaranteed to
be delivered before the function ends.
.. code-block:: python
@with_sentry(
config=SentryConfig(
dsn="SomeDSN",
env="production",
tags={"service": "etl-pipeline"},
),
runtime_config=SentryRuntimeConfig(
capture_exceptions=True,
flush=True,
tags={"job": "daily-sync"},
extra={"rows_processed": 9999},
),
)
def run_etl():
pass
..
:param _func: Internal that allows the decorator to be used without parentheses.
:param config: Sentry SDK configuration. See :class:`~core_sentry.base.SentryConfig`.
:param runtime_config:
Decorator runtime behavior,exception capture, flush,
and per-event tags/extra. See :class:`~core_sentry.base.SentryRuntimeConfig`.
Defaults to ``SentryRuntimeConfig()`` (all flags off, no tags/extra).
.. note::
Both flags default to ``False`` because the Sentry SDK already captures
unhandled exceptions via ``sys.excepthook`` and flushes via an ``atexit``
handler. Set them to ``True`` for ETLs and short-lived scripts running
in containers, where a SIGKILL or OOM kill can prevent ``atexit`` from
running and where per-invocation tags and extra context matter.
"""
def decorator(func: typing.Callable) -> typing.Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
_init_sentry(config)
try:
result = func(*args, **kwargs)
except Exception as exc:
if runtime_config.capture_exceptions:
_capture_exception(
exc,
tags=runtime_config.tags,
extra=runtime_config.extra,
)
raise
finally:
if runtime_config.flush:
_flush_sentry(timeout=runtime_config.flush_timeout)
return result
return wrapper
if _func is None:
return decorator
return decorator(_func)