Core Primitives#

Configuration, SDK initialization, and the main event-lifecycle functions. These are the building blocks used directly or by the decorator layer.

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:

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)

class core_sentry.base.SentryConfig(dsn: str, env: str, tags: Dict[str, str] | None=None, traces_sample_rate: float = 0.1, options: Dict[str, ~typing.Any]=<factory>)[source]#

Bases: object

Configuration for Sentry SDK initialization.

Parameters:

dsn – The DSN tells the SDK where to send the events.
env – Environments tell you where an error occurred (production, staging).
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/
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.
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: Dict[str, str] | None = None#

traces_sample_rate: float = 0.1#

options: Dict[str, Any]#

__init__(dsn: str, env: str, tags: Dict[str, str] | None=None, traces_sample_rate: float = 0.1, options: Dict[str, ~typing.Any]=<factory>) → None#

core_sentry.base.init_sentry(config: SentryConfig) → None[source]#

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.

Parameters:: 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,
))

core_sentry.base.flush_sentry(timeout: float = 2.0) → bool[source]#

Flush pending Sentry events. I should be called before process exit in scripts/ETLs.

Parameters:: 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.")

core_sentry.base.capture_exception(error: BaseException, tags: Dict[str, str] | None = None, extra: Dict[str, Any] | None = None) → str | None[source]#

Capture an exception and send it to Sentry.

Parameters:

error – Exception to capture.
tags – Per-event tags (do not affect subsequent captures).
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"},
    )

core_sentry.base.capture_message(message: str, level: Literal['fatal', 'critical', 'error', 'warning', 'info', 'debug'] = 'info', tags: Dict[str, str] | None = None, extra: Dict[str, Any] | None = None) → str | None[source]#

Capture a message and send it to Sentry.

Parameters:

message – Message text to send.
level – Sentry severity level (“debug”, “info”, “warning”, “error”, “fatal”).
tags – Per-event tags (do not affect subsequent captures).
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"},
)

Core Primitives#

This Page