`alembic_pg_autogen.canonicalize`¶

Desired-state canonicalization through PostgreSQL round-tripping.

Module Contents¶

Classes¶

CanonicalState

Post-DDL catalog snapshot returned by canonicalize().

Functions¶

`canonicalize`	Canonicalize user-provided DDL by round-tripping through PostgreSQL.
`canonicalize_functions`	Canonicalize function DDL and return the resulting `FunctionInfo` list.
`canonicalize_triggers`	Canonicalize trigger DDL and return the resulting `TriggerInfo` list.

Data¶

log

API¶

alembic_pg_autogen.canonicalize.log = 'getLogger(...)'¶

class alembic_pg_autogen.canonicalize.CanonicalState¶

Bases: typing.NamedTuple

Post-DDL catalog snapshot returned by canonicalize().

functions: collections.abc.Sequence[alembic_pg_autogen.inspect.FunctionInfo] = None¶

triggers: collections.abc.Sequence[alembic_pg_autogen.inspect.TriggerInfo] = None¶

alembic_pg_autogen.canonicalize.canonicalize(conn: sqlalchemy.Connection, *, function_ddl: collections.abc.Sequence[str] = (), trigger_ddl: collections.abc.Sequence[str] = (), schemas: collections.abc.Sequence[str] | None = None) → alembic_pg_autogen.canonicalize.CanonicalState¶

Canonicalize user-provided DDL by round-tripping through PostgreSQL.

Executes the given DDL statements inside a savepoint, reads back canonical forms via inspect_functions / inspect_triggers, then rolls back the savepoint — leaving the database unchanged.

Function DDL is executed before trigger DDL so that triggers may reference functions declared in the same batch.

Args:: conn: An open SQLAlchemy connection (may have an active transaction). function_ddl: CREATE FUNCTION / CREATE PROCEDURE statements. trigger_ddl: CREATE TRIGGER statements. schemas: Optional schema list passed to the inspect helpers. When None, all user schemas are included.
Returns:: A CanonicalState with the full post-DDL catalog state.
Raises:: sqlalchemy.exc.DBAPIError: If any DDL statement is invalid.

alembic_pg_autogen.canonicalize.canonicalize_functions(conn: sqlalchemy.Connection, ddl: collections.abc.Sequence[str], schemas: collections.abc.Sequence[str] | None = None) → collections.abc.Sequence[alembic_pg_autogen.inspect.FunctionInfo]¶

Canonicalize function DDL and return the resulting FunctionInfo list.

Convenience wrapper around canonicalize() with only function_ddl populated.

alembic_pg_autogen.canonicalize.canonicalize_triggers(conn: sqlalchemy.Connection, ddl: collections.abc.Sequence[str], schemas: collections.abc.Sequence[str] | None = None) → collections.abc.Sequence[alembic_pg_autogen.inspect.TriggerInfo]¶

Canonicalize trigger DDL and return the resulting TriggerInfo list.

Convenience wrapper around canonicalize() with only trigger_ddl populated.

alembic_pg_autogen.canonicalize¶