Skip to content

baldur.adapters.sql — Framework-Free SQL Adapter (DB-API 2.0)

Generic repository base plus priority-1 SQL-backed implementations of Baldur's core repositories. Works with any DB-API 2.0 driver (psycopg2, mysql-connector-python, stdlib sqlite3) — selected by DSN scheme.

sql

Framework-free SQL adapter (DB-API 2.0).

Provides a generic repository base plus priority-1 SQL-backed implementations of Baldur's core repositories. Works with any DB-API 2.0 driver — psycopg2, mysql-connector-python, or the stdlib sqlite3 — selected by DSN scheme.

Status: Public

GenericSQLRepository 

GenericSQLRepository(
    get_connection: Callable[[], Any],
    *,
    dialect: SQLDialect | None = None,
    autocommit_delegated: bool | None = None,
    schema: (
        tuple[str, int, Callable[[SQLDialect], list[str]]]
        | None
    ) = None
)

DB-API 2.0 repository helpers.

Subclasses inherit this class plus a domain-specific ABC (FailedOperationRepository etc.). The base exposes helpers only — no ABC method has a default implementation here, so there is no MRO ambiguity.

Connection ownership: get_connection is a user-supplied callable. Baldur does not own a pool. Each helper borrows a connection via the callable and relies on the callable's close() / return-to-pool semantics. Common implementations:

  • get_connection = lambda: psycopg2.connect(DSN) — direct.
  • get_connection = engine.raw_connection — SQLAlchemy pool.
  • get_connection = lambda: pgbouncer_pool.getconn() — external pooler.

SchemaVersionManager 

SchemaVersionManager(
    get_connection: Callable[[], Any],
    *,
    dialect: SQLDialect
)

Owns the baldur_schema_version bookkeeping table.

Repos call ensure(repo_name, version, ddl_statements) during first use. DDL runs exactly once per (repo_name, version) pair per database — subsequent calls are no-ops.

SQLEventJournalRepository 

SQLEventJournalRepository(
    get_connection: Callable[[], Any],
    *,
    dialect: SQLDialect | None = None,
    autocommit_delegated: bool | None = None,
    max_query_limit: int = 10000
)

Bases: GenericSQLRepository, EventJournalRepository

DB-API 2.0 backed append-only event journal.

SQLFailedOperationRepository 

SQLFailedOperationRepository(
    get_connection: Callable[[], Any],
    *,
    dialect: SQLDialect | None = None,
    autocommit_delegated: bool | None = None
)

Bases: GenericSQLRepository, FailedOperationRepository

DB-API 2.0 backed DLQ repository.

count_created_in_window 

count_created_in_window(
    start: datetime, end: datetime
) -> int

Count rows whose created_at is in the inclusive [start, end].

Backed by idx_baldur_dlq_created_at — a range seek, not a scan.

find_replayable_page 

find_replayable_page(
    *,
    max_retries: int,
    domain: str | None = None,
    failure_type: str | None = None,
    source: str | None = None,
    limit: int = 100,
    cursor: str | None = None
) -> ReplayablePage

Keyset-paged replayable selection over idx_baldur_dlq_status_domain.

The seek binds a rebuilt datetime through _dt_to_db, never a rendered timestamp: sqlite stores created_at as TEXT in the driver's own space-separated form, which sorts below the ISO T form, so a string comparison silently matches nothing — and an empty page is exactly how a caller learns the queue is drained.

source lives inside the JSON payload, so it is the one residual applied in Python; the loop keeps fetching windows until it has limit matches or has examined REPLAY_SELECTION_MAX_SCAN rows. Each window re-seeks from the row the previous one ended on, so a row another drainer resolves mid-walk cannot shift the members behind it past the next window's start.

get_facet_counts 

get_facet_counts(
    *, status: str | None = None, domain: str | None = None
) -> dict[str, dict[str, int]]

Faceted status×domain counts via GROUP BY.

by_status is scoped by domain; by_domain is scoped by status. GROUP BY drops zero-count buckets structurally. The domain-scoped by_status (WHERE domain GROUP BY status) is a covering scan over idx_baldur_dlq_status_domain; the status-scoped by_domain (WHERE status GROUP BY domain) is a prefix seek on the same composite. Both exact, both fine on the cold operator read path.

try_acquire_for_replay 

try_acquire_for_replay(
    id: str, max_retries: int, force: bool = False
) -> FailedOperationData | None

Atomically flip PENDING → REPLAYING if retry budget remains.

Uses a conditional UPDATE as the concurrency guard. cursor.rowcount

0 means this worker won the race; the row is then re-read within the same transaction so no other writer can delete or mutate it between the claim and the DTO return.

force=True is the operator cap-override: it widens the WHERE status set to {PENDING, REQUIRES_REVIEW}, drops the retry_count < max_retries bound, resets retry_count to a fresh budget (1), and stamps the metadata history scar into the JSON payload — all inside the same transaction so the SELECT→UPDATE claim stays race-free against a concurrent sweep. See FailedOperationRepository.try_acquire_for_replay.

get_compressed_entry 

get_compressed_entry(
    entry_id: str,
) -> DLQCompressedEntry | None

Return a single compressed entry by id, or None if absent.

A primary-key point read on the id VARCHAR column — proportional to matches, like every SQL compressed query.

get_compressed_entries_before 

get_compressed_entries_before(
    *,
    status: str,
    before: datetime,
    limit: int = 100,
    offset: int = 0,
    after: datetime | None = None
) -> list[DLQCompressedEntry]

Query compressed entries in a cutoff window, oldest first.

The (status, compressed_at) index serves both bounds and the order.

update_compressed_status 

update_compressed_status(
    entry_id: str, new_status: str
) -> bool

Transition compressed entry lifecycle status.

Stamps the matching timestamp column alongside the status, as the memory and Redis adapters do. STALE -> ARCHIVED is driven off stale_at, so an unstamped transition would strand the entry.

SQLRecoverySessionArchiveRepository 

SQLRecoverySessionArchiveRepository(
    get_connection: Callable[[], Any],
    *,
    dialect: SQLDialect | None = None,
    autocommit_delegated: bool | None = None
)

Bases: GenericSQLRepository, RecoverySessionArchiveRepository

DB-API 2.0 backed recovery session archive repository.

SQLSecurityIncidentRepository 

SQLSecurityIncidentRepository(
    get_connection: Callable[[], Any],
    *,
    dialect: SQLDialect | None = None,
    autocommit_delegated: bool | None = None
)

Bases: GenericSQLRepository, SecurityIncidentRepository

DB-API 2.0 backed security incident repository.

SQLStatisticsRepository 

SQLStatisticsRepository(
    get_connection: Callable[[], Any],
    *,
    dialect: SQLDialect | None = None,
    autocommit_delegated: bool | None = None
)

Bases: GenericSQLRepository, StatisticsRepositoryInterface

DB-API 2.0 backed statistics repository.

Reads the baldur_dlq table directly and does not own a table — schema bootstrap is skipped. The circuit-breaker methods read a table Baldur never writes (CB state stays in memory or Redis) and return empty results when it is absent.

sql_transaction 

sql_transaction(conn: Any) -> Any

Suspend repo-scoped auto-commit for the duration of the block.

Usage::

with sql_transaction(conn):
    dlq_repo.create(...)
    postmortem_repo.save(...)
# single commit (or rollback on exception) applies to both.

All repositories whose get_connection returns conn during the block skip their per-call commit. The context manager itself issues the final commit, or rollback on exception.