orxtra v0.13.0 /services.src.orxtra.services._schema
Updated
On this page

AsyncpgAdapter/AsyncpgTx bridge asyncpg to the generated schema executor's protocols; verify_schema() runs its verify() and raises SchemaError with an actionable message if DB objects are missing.

#services.src.orxtra.services._schema

#services.src.orxtra.services._schema

Shared schema verification and asyncpg adapter for the generated executor.

Provides:

  • AsyncpgAdapter / AsyncpgTx: bridge between asyncpg's native types

and the generated schema executor's AsyncConnection / AsyncTransaction protocols. Previously duplicated in cli/_db.py, tests/pg_fixtures.py, and tests/test_db_commands.py.

  • SchemaError: raised when the database schema is incomplete.
  • verify_schema(pool): single importable function that runs the generated

executor's verify() and raises SchemaError with an actionable message if any required objects are missing. Designed for use by every long-running DB-backed process: orxtra serve, CLI commands, dispatcher worker, incoming receiver.

The helper filters known false positives from verify():

  • comments section: pg_catalog has no query for COMMENT ON statements.
  • Functions and triggers that the executor places in the indexes section

(the existence checker queries pg_indexes, which only has actual indexes).

#AsyncpgTx

Adapter wrapping asyncpg transaction to satisfy AsyncTransaction.

#execute

python
async def execute(self, query: str) -> None

#AsyncpgAdapter

Adapter wrapping asyncpg.Connection to satisfy AsyncConnection.

#execute

python
async def execute(self, query: str) -> None

#fetch

python
async def fetch(self, query: str) -> list[dict[str, Any]]

#transaction

python
def transaction(self) -> AsyncpgTx

#_is_false_positive

python
def _is_false_positive(kind: str, name: str) -> bool

Return True if this missing entry is a known false positive.

The generated executor places functions (pgdesign_deny_mutation) and triggers (*.deny_mutation) in the "indexes" section. The existence checker for "indexes" queries pg_indexes, which only contains actual indexes -- so functions and triggers always appear missing.

#SchemaError

Raised when the database schema is incomplete or outdated.

#verify_schema_objects

python
async def verify_schema_objects(adapter: AsyncpgAdapter) -> tuple[list[tuple[str, str]], list[tuple[str, str]]]

Run the generated verify() with known false positives filtered.

Runs the generated executor's verify() excluding the sections and entries that have no reliable existence check (see _VERIFY_EXCLUDE_SECTIONS and _is_false_positive), so callers get a truthful present/missing split.

Args:

  • adapter: An AsyncpgAdapter wrapping a live connection.

Returns:

  • A (present, real_missing) tuple of (kind, name) lists.

#verify_schema

python
async def verify_schema(pool: asyncpg.Pool[Any]) -> None

Verify the database schema is complete, raise on missing objects.

Acquires a single connection from the pool, runs the generated executor's verify(), filters known false positives, and raises SchemaError with an actionable message if real objects are missing.

Returns silently if everything is present.

Args:

  • pool: An asyncpg connection pool connected to the target database.

Raises:

  • SchemaError: If required schema objects are missing.
Search