Schema

Schema definitions, validation, and migration.

PostgreSQL schema migration runner.

A minimal, dependency-light tool for applying versioned raw-SQL DDL migrations to a PostgreSQL database. Domain-agnostic — consumers supply their own DSN, migrations directory, and tracking-table location.

Does NOT compete with Django migrations. Its tracking table lives in a consumer-configured schema, independent of django_migrations. The two systems can coexist on the same database so long as they manage different schemas / tables.

Typical usage from a consumer project:

from siege_utilities.schema import MigrationRunner

runner = MigrationRunner(
    dsn=os.environ["DATABASE_URL"],
    migrations_dir="schema/migrations/",
    tracking_schema="myapp",
    tracking_table="_schema_migrations",
)
runner.apply_all()

See MigrationRunner for the full API.

class siege_utilities.schema.MigrationRunner[source]

Bases: object

Apply versioned SQL migrations to a PostgreSQL database.

Parameters:

dsn – Standard PostgreSQL DSN string (postgresql://user:pass@host:port/db).
migrations_dir – Path to the directory containing V*__*.sql files.
tracking_schema – Schema where the tracking table lives. Must be a valid SQL identifier. Typically consumer-specific (e.g., "myapp"). The schema is created if it doesn’t exist.
tracking_table – Name of the tracking table. Must be a valid SQL identifier. Defaults to "_schema_migrations".

Raises:

RuntimeError – if psycopg is not installed.
ValueError – if tracking_schema or tracking_table is not a valid SQL identifier.

__init__(*, dsn, migrations_dir, tracking_schema, tracking_table='_schema_migrations')[source]

Parameters:

dsn (str)
migrations_dir (str | Path)
tracking_schema (str)
tracking_table (str)

Return type:

None

discover_migrations()[source]

Read all migration files from migrations_dir, sorted by version.

Returns a list in lexical version order. Raises FileNotFoundError if the directory is missing; returns [] if it’s empty.

Return type:: list[MigrationFile]

applied_migrations()[source]

Return rows from the tracking table, oldest first.

If the tracking table doesn’t exist yet, returns an empty list.

Return type:: list[AppliedMigration]

status()[source]

Return (version, status) for every discovered-or-applied migration.

Includes applied migrations that are no longer on disk (orphans), which typically indicate a concerning drift between the DB and the code.

Return type:: list[tuple[str, MigrationStatus]]

pending_migrations()[source]

Return migrations on disk that are not in the tracking table.

Return type:: list[PendingMigration]

verify_checksums()[source]

Compare on-disk checksum vs tracking-table checksum for every applied migration.

Returns a list of (version, disk_checksum, tracked_checksum) for any migrations whose on-disk content has drifted from what was originally applied. Empty list = healthy. A non-empty result is a warning that migration files have been edited after application — a correctness hazard.

Return type:: list[tuple[str, str, str]]

apply_all(*, _pending=None)[source]

Apply every pending migration in version order.

Returns the list of migrations that were applied during this call.

Parameters:: _pending (list[PendingMigration] | None) – Optional pre-discovered pending list. Pass the result of a prior pending_migrations() call to let _apply_one’s divergence guard detect mid-flight file mutations. When None (the default), pending_migrations() is called internally. This parameter is primarily intended for testing.
Return type:: list[PendingMigration]

Concurrency: This method is intended for single-process use. There is a TOCTOU gap between pending_migrations() and the per-migration apply: a second concurrent caller that sees the same pending set will attempt to apply the same migrations and fail on the version PRIMARY KEY constraint. Use an external serialization mechanism (OS-level lock file, advisory lock, deploy-time mutex) if concurrent callers are possible.

apply_up_to(target_version)[source]

Apply pending migrations in order, stopping once target_version is applied.

Returns an empty list if target_version is already applied. Raises ValueError if target_version is not on disk.

Concurrency: Same single-process caveat as apply_all().

Parameters:: target_version (str)
Return type:: list[PendingMigration]

class siege_utilities.schema.MigrationFile[source]

Bases: object

A migration file on disk.

version: str

description: str

path: Path

sql_text: str

checksum: str

classmethod from_path(path)[source]

Parse V<version>__<description>.sql and load its SQL body.

Raises:: ValueError – If the filename does not match the expected pattern.
Parameters:: path (Path)
Return type:: MigrationFile

__init__(version, description, path, sql_text, checksum)

Parameters:

version (str)
description (str)
path (Path)
sql_text (str)
checksum (str)

Return type:

None

class siege_utilities.schema.MigrationStatus[source]

Bases: Enum

Result of inspecting the tracking table against on-disk migrations.

APPLIED = 'applied'

PENDING = 'pending'

class siege_utilities.schema.AppliedMigration[source]

Bases: object

A migration row read from the tracking table.

version: str

description: str

checksum: str

applied_at: datetime

__init__(version, description, checksum, applied_at)

Parameters:

version (str)
description (str)
checksum (str)
applied_at (datetime)

Return type:

None

class siege_utilities.schema.PendingMigration[source]

Bases: object

A migration file that has not yet been applied to the database.

version: str

description: str

checksum: str

path: Path

__init__(version, description, checksum, path)

Parameters:

version (str)
description (str)
checksum (str)
path (Path)

Return type:

None

Submodules

PostgreSQL schema migration runner.

Applies numbered raw-SQL migration files in order, tracking applied versions in a consumer-configured tracking table. Each migration is applied inside its own transaction together with the tracking-table insert, so a failed migration leaves no partial state behind.

Conventions:

Migration files live in a single directory.
Filenames match V<version>__<description>.sql — the version string is the primary key (any string that sorts correctly works; zero-padded numerics like 0001 are recommended).
Tracking schema and table are consumer-configured so multiple projects can share a database without colliding.
The runner uses psycopg (v3). Callers supply a standard PostgreSQL DSN string.

The runner does NOT support down-migrations. Schema rollback is out of scope — recovery is a restore-from-backup or a forward-only corrective migration. Consumers who need rollbacks should reach for a heavier tool.

class siege_utilities.schema.migration_runner.AppliedMigration[source]

Bases: object

A migration row read from the tracking table.

version: str

description: str

checksum: str

applied_at: datetime

__init__(version, description, checksum, applied_at)

Parameters:

version (str)
description (str)
checksum (str)
applied_at (datetime)

Return type:

None

class siege_utilities.schema.migration_runner.MigrationFile[source]

Bases: object

A migration file on disk.

version: str

description: str

path: Path

sql_text: str

checksum: str

classmethod from_path(path)[source]

Parse V<version>__<description>.sql and load its SQL body.

Raises:: ValueError – If the filename does not match the expected pattern.
Parameters:: path (Path)
Return type:: MigrationFile

__init__(version, description, path, sql_text, checksum)

Parameters:

version (str)
description (str)
path (Path)
sql_text (str)
checksum (str)

Return type:

None

class siege_utilities.schema.migration_runner.MigrationRunner[source]

Bases: object

Apply versioned SQL migrations to a PostgreSQL database.

Parameters:

dsn – Standard PostgreSQL DSN string (postgresql://user:pass@host:port/db).
migrations_dir – Path to the directory containing V*__*.sql files.
tracking_schema – Schema where the tracking table lives. Must be a valid SQL identifier. Typically consumer-specific (e.g., "myapp"). The schema is created if it doesn’t exist.
tracking_table – Name of the tracking table. Must be a valid SQL identifier. Defaults to "_schema_migrations".

Raises:

RuntimeError – if psycopg is not installed.
ValueError – if tracking_schema or tracking_table is not a valid SQL identifier.

__init__(*, dsn, migrations_dir, tracking_schema, tracking_table='_schema_migrations')[source]

Parameters:

dsn (str)
migrations_dir (str | Path)
tracking_schema (str)
tracking_table (str)

Return type:

None

discover_migrations()[source]

Read all migration files from migrations_dir, sorted by version.

Returns a list in lexical version order. Raises FileNotFoundError if the directory is missing; returns [] if it’s empty.

Return type:: list[MigrationFile]

applied_migrations()[source]

Return rows from the tracking table, oldest first.

If the tracking table doesn’t exist yet, returns an empty list.

Return type:: list[AppliedMigration]

status()[source]

Return (version, status) for every discovered-or-applied migration.

Includes applied migrations that are no longer on disk (orphans), which typically indicate a concerning drift between the DB and the code.

Return type:: list[tuple[str, MigrationStatus]]

pending_migrations()[source]

Return migrations on disk that are not in the tracking table.

Return type:: list[PendingMigration]

verify_checksums()[source]

Compare on-disk checksum vs tracking-table checksum for every applied migration.

Returns a list of (version, disk_checksum, tracked_checksum) for any migrations whose on-disk content has drifted from what was originally applied. Empty list = healthy. A non-empty result is a warning that migration files have been edited after application — a correctness hazard.

Return type:: list[tuple[str, str, str]]

apply_all(*, _pending=None)[source]

Apply every pending migration in version order.

Returns the list of migrations that were applied during this call.

Parameters:: _pending (list[PendingMigration] | None) – Optional pre-discovered pending list. Pass the result of a prior pending_migrations() call to let _apply_one’s divergence guard detect mid-flight file mutations. When None (the default), pending_migrations() is called internally. This parameter is primarily intended for testing.
Return type:: list[PendingMigration]

Concurrency: This method is intended for single-process use. There is a TOCTOU gap between pending_migrations() and the per-migration apply: a second concurrent caller that sees the same pending set will attempt to apply the same migrations and fail on the version PRIMARY KEY constraint. Use an external serialization mechanism (OS-level lock file, advisory lock, deploy-time mutex) if concurrent callers are possible.

apply_up_to(target_version)[source]

Apply pending migrations in order, stopping once target_version is applied.

Returns an empty list if target_version is already applied. Raises ValueError if target_version is not on disk.

Concurrency: Same single-process caveat as apply_all().

Parameters:: target_version (str)
Return type:: list[PendingMigration]

class siege_utilities.schema.migration_runner.MigrationStatus[source]

Bases: Enum

Result of inspecting the tracking table against on-disk migrations.

APPLIED = 'applied'

PENDING = 'pending'

class siege_utilities.schema.migration_runner.PendingMigration[source]

Bases: object

A migration file that has not yet been applied to the database.

version: str

description: str

checksum: str

path: Path

__init__(version, description, checksum, path)

Parameters:

version (str)
description (str)
checksum (str)
path (Path)

Return type:

None