Skip to content

baldur.interfaces — PostgreSQL Admin

The PostgreSQL-admin provider contract — connection statistics and advisory-lock results.

PostgreSQL admin

PgAdminProvider

Bases: ABC

Abstract interface for PostgreSQL admin SQL primitives.

Concrete implementations route SQL through a runtime-specific cursor (Django connections[alias].cursor(), DB-API 2.0 conn.cursor(), or noop). Callers should not assume any specific backend.

Availability gate: is_available() returns False for the no-op implementation so consumers can omit PG-specific keys from their response dicts when the underlying runtime cannot satisfy the contract.

is_available abstractmethod 

is_available() -> bool

Return True iff this provider can execute PG admin SQL.

ping abstractmethod 

ping() -> bool

Verify connectivity with SELECT 1.

get_connection_stats abstractmethod 

get_connection_stats() -> ConnectionStats

Read connection state counts from pg_stat_activity.

get_active_connection_count abstractmethod 

get_active_connection_count() -> int

Return the count of state = 'active' rows in pg_stat_activity.

pg_sleep abstractmethod 

pg_sleep(seconds: float) -> None

Execute SELECT pg_sleep(...).

execute_slow_query abstractmethod 

execute_slow_query(seconds: int) -> None

Hold a backend session via pg_sleep for the given duration.

get_backend_pid_with_delay abstractmethod 

get_backend_pid_with_delay(
    delay_seconds: float = 0.01,
) -> int

Return pg_backend_pid() after a brief pg_sleep delay.

acquire_advisory_lock abstractmethod 

acquire_advisory_lock(
    lock_id: int, wait: bool = True
) -> bool

Acquire an exclusive advisory lock.

acquire_advisory_lock_shared abstractmethod 

acquire_advisory_lock_shared(
    lock_id: int, wait: bool = True
) -> bool

Acquire a shared advisory lock.

release_advisory_lock abstractmethod 

release_advisory_lock(lock_id: int) -> bool

Release an exclusive advisory lock.

release_advisory_lock_shared abstractmethod 

release_advisory_lock_shared(lock_id: int) -> bool

Release a shared advisory lock.

try_advisory_lock abstractmethod 

try_advisory_lock(lock_id: int) -> bool

Attempt a non-blocking exclusive advisory lock.

set_lock_timeout abstractmethod 

set_lock_timeout(timeout_ms: int) -> None

Set session-level lock_timeout in milliseconds (0 = unlimited).

set_statement_timeout abstractmethod 

set_statement_timeout(timeout_ms: int) -> None

Set session-level statement_timeout in milliseconds (0 = unlimited).

reset_timeouts abstractmethod 

reset_timeouts() -> None

Reset both lock_timeout and statement_timeout to unlimited.

execute_aggregate_query abstractmethod 

execute_aggregate_query(
    table_name: str,
) -> tuple[int, float, float, float]

Execute COUNT/AVG/MAX/MIN over table_name.

Returns (total_count, avg_price, max_price, min_price).

execute_nonexistent_table_query abstractmethod 

execute_nonexistent_table_query() -> None

Execute a query against a missing table (used to trigger CB failures).

execute_timeout_query abstractmethod 

execute_timeout_query(
    timeout_ms: int = 1, sleep_seconds: int = 1
) -> None

Execute a query that exceeds statement_timeout (CB testing).

advisory_lock_context abstractmethod 

advisory_lock_context(
    lock_id: int, exclusive: bool = True, wait: bool = True
) -> Generator[bool, None, None]

Hold an advisory lock for the duration of a with block.

timeout_context abstractmethod 

timeout_context(
    lock_timeout_ms: int = 0, statement_timeout_ms: int = 0
) -> Generator[None, None, None]

Apply session timeouts within a with block, restoring on exit.

create_cursor abstractmethod 

create_cursor() -> Any

Create a cursor whose lifecycle is owned by the caller.

Used by pool-exhaustion paths that hold cursors externally (StressTestService._held_connections).

execute_with_cursor abstractmethod 

execute_with_cursor(
    cursor: Any, query: str, params: list[Any] | None = None
) -> Any

Execute query on a caller-supplied cursor and return fetchone().

ConnectionStats dataclass 

ConnectionStats(
    total_connections: int,
    active: int,
    idle: int,
    idle_in_transaction: int,
)

PostgreSQL connection statistics from pg_stat_activity.

AdvisoryLockResult dataclass 

AdvisoryLockResult(
    acquired: bool, lock_id: int, error: str | None = None
)

Result of an advisory lock attempt.