orxtra v0.13.0 /services.src.orxtra.services._generated.schema_executor
Updated
On this page

API reference for the services.src.orxtra.services._generated.schema_executor module — auto-generated documentation covering public functions, classes, and type signatures.

#services.src.orxtra.services._generated.schema_executor

#services.src.orxtra.services._generated.schema_executor

#AsyncConnection

#execute

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

#fetch

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

#transaction

python
async def transaction(self) -> AsyncTransaction

#AsyncTransaction

#execute

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

#Section

#exists

python
async def exists(self, conn: AsyncConnection, op: DDLOp) -> bool

Check if the object described by op already exists.

#ExecutionResult

ExecutionResult
FieldTypeDefault
executedlist[tuple[str, str]]field(default_factory=list)
skippedlist[tuple[str, str]]field(default_factory=list)
errorslist[tuple[str, str, str]]field(default_factory=list)

#VerifyResult

VerifyResult
FieldTypeDefault
presentlist[tuple[str, str]]field(default_factory=list)
missinglist[tuple[str, str]]field(default_factory=list)

#execute

python
async def execute(conn: AsyncConnection, sections: Sequence[str] | None=None, exclude_sections: Sequence[str] | None=None, idempotent: bool=True, dry_run: bool=False, extension_stubs: dict[str, str] | None=None) -> ExecutionResult

Execute DDL sections.

Two-phase execution: transactional sections run inside a single transaction, non-transactional sections run outside afterward.

Ops with idempotent_sql=None are inherently re-runnable statements (COMMENT ON, SET STATISTICS, OWNER TO, ENABLE/FORCE ROW LEVEL SECURITY, partman config UPDATE); for those, executing op.sql again is harmless, so idempotent=True falls back to op.sql. Exception: partman create_parent has no idempotent form and errors if re-run.

#verify

python
async def verify(conn: AsyncConnection, sections: Sequence[str] | None=None, exclude_sections: Sequence[str] | None=None) -> VerifyResult

Check which ops across selected sections exist in the database.

#create_schema

python
async def create_schema(conn: AsyncConnection) -> ExecutionResult

Execute all DDL, failing on conflicts.

#ensure_schema

python
async def ensure_schema(conn: AsyncConnection) -> ExecutionResult

Execute all DDL idempotently, safe to run on existing databases.

Search