alembic_pg_autogen.compare

Comparator functions for detecting PostgreSQL object diffs during autogenerate.

Module Contents

Classes

_HasText	Protocol for objects with a .text string attribute (e.g. SQLAlchemy TextClause).
SQLCreatable	Protocol for alembic-utils-style objects that produce DDL via to_sql_statement_create().

Functions

setup	Register the PostgreSQL object comparator with Alembic’s plugin system.
_compare_pg_objects	Compare current database state against desired functions/triggers.
_resolve_ddl	Convert a mixed sequence of DDL strings and SQLCreatable objects to plain DDL strings.
_filter_to_declared	Filter canonical state to only include objects declared in user DDL.
_parse_function_names	Extract (schema, name) pairs from function DDL strings via postgast.
_parse_trigger_identities	Extract (schema, table_name, trigger_name) triples from trigger DDL strings via postgast.
_get_default_schema	Get the current schema for the connection.
_resolve_schemas	Convert Alembic’s schema set to a list suitable for inspect functions.
_filter_to_schemas	Filter a CanonicalState to only include objects in the given schemas.
_order_ops	Convert diff ops to MigrateOperation instances in dependency-safe order.

Data

log

API

alembic_pg_autogen.compare.log = 'getLogger(...)'

class alembic_pg_autogen.compare._HasText

Bases: typing.Protocol

Protocol for objects with a .text string attribute (e.g. SQLAlchemy TextClause).

property text: str

class alembic_pg_autogen.compare.SQLCreatable

Bases: typing.Protocol

Protocol for alembic-utils-style objects that produce DDL via to_sql_statement_create().

Any object implementing this method (e.g. PGFunction, PGView, PGTrigger from alembic-utils) can be passed wherever this library expects DDL strings. The DDL is extracted by calling obj.to_sql_statement_create().text.

to_sql_statement_create() → alembic_pg_autogen.compare._HasText

alembic_pg_autogen.compare.setup(plugin: alembic.runtime.plugins.Plugin) → None

Register the PostgreSQL object comparator with Alembic’s plugin system.

alembic_pg_autogen.compare._compare_pg_objects(autogen_context: alembic.autogenerate.api.AutogenContext, upgrade_ops: alembic.operations.ops.UpgradeOps, schemas: set[str | None]) → alembic.util.PriorityDispatchResult

Compare current database state against desired functions/triggers.

alembic_pg_autogen.compare._resolve_ddl(items: collections.abc.Sequence[str | alembic_pg_autogen.compare.SQLCreatable]) → tuple[str, ...]

Convert a mixed sequence of DDL strings and SQLCreatable objects to plain DDL strings.

Strings are passed through unchanged. SQLCreatable objects (e.g. alembic-utils entities) are converted by calling obj.to_sql_statement_create().text.

alembic_pg_autogen.compare._filter_to_declared(canonical: alembic_pg_autogen.canonicalize.CanonicalState, pg_functions: collections.abc.Sequence[str], pg_triggers: collections.abc.Sequence[str], conn: sqlalchemy.Connection) → alembic_pg_autogen.canonicalize.CanonicalState

Filter canonical state to only include objects declared in user DDL.

canonicalize() returns a full catalog snapshot after executing DDL, which includes pre-existing objects. This function parses identity info from the raw DDL strings and keeps only those canonical entries that match.

alembic_pg_autogen.compare._parse_function_names(ddl_list: collections.abc.Sequence[str], conn: sqlalchemy.Connection) → set[tuple[str, str]]

Extract (schema, name) pairs from function DDL strings via postgast.

Raises:

ValueError: If any DDL string does not contain a valid CREATE FUNCTION statement.

alembic_pg_autogen.compare._parse_trigger_identities(ddl_list: collections.abc.Sequence[str], conn: sqlalchemy.Connection) → set[tuple[str, str, str]]

Extract (schema, table_name, trigger_name) triples from trigger DDL strings via postgast.

Raises:

ValueError: If any DDL string does not contain a valid CREATE TRIGGER statement.

alembic_pg_autogen.compare._get_default_schema(conn: sqlalchemy.Connection) → str

Get the current schema for the connection.

alembic_pg_autogen.compare._resolve_schemas(conn: sqlalchemy.Connection, schemas: collections.abc.Iterable[str | None]) → list[str] | None

Convert Alembic’s schema set to a list suitable for inspect functions.

Alembic passes {None} to mean “only the default schema”. This function resolves None to the connection’s current_schema() so inspect functions receive concrete schema names. If the resulting set covers all user schemas, returns None (meaning “no filter”).

alembic_pg_autogen.compare._filter_to_schemas(state: alembic_pg_autogen.canonicalize.CanonicalState, schemas: collections.abc.Iterable[str] | None) → alembic_pg_autogen.canonicalize.CanonicalState

Filter a CanonicalState to only include objects in the given schemas.

alembic_pg_autogen.compare._order_ops(function_ops: collections.abc.Sequence[alembic_pg_autogen.diff.FunctionOp], trigger_ops: collections.abc.Sequence[alembic_pg_autogen.diff.TriggerOp]) → list[alembic.operations.ops.MigrateOperation]

Convert diff ops to MigrateOperation instances in dependency-safe order.

Order: drop triggers, drop functions, create/replace functions, create/replace triggers.