Migration Runner¶

iceberg_ops.runner¶

MigrationRunner — discovers, applies, and rolls back Iceberg migrations for a specific namespace inside a catalog.

The runner manages one namespace at a time. To manage multiple namespaces (e.g. "sales" and "finance"), create one runner per namespace — they share the same catalog and the same history table but track state independently via the namespace column in _iceberg_ops.migration_history.

Migration directory layout¶

The runner supports two directory layouts:

Flat (default, backwards compatible):

migrations/
  0001_create_orders.py
  0002_add_sku.py

Namespace-scoped (preferred for multi-namespace projects):
```
migrations/
  sales/
    0001_create_orders.py
  finance/
    0001_create_invoices.py
```
When the runner’s namespace matches a subdirectory name under migrations_dir, that subdirectory is used automatically. If no matching subdirectory exists, the flat layout is used as a fallback.

Concurrent runs¶

Concurrent runs are not guarded. Two migrate runs for one namespace can both execute upgrade(); the second run to record history raises ConcurrentMigrationError and history keeps one row per (version, namespace); its upgrade() may already have run, so upgrade() must be idempotent. Run migrations for a namespace from one job at a time (CI concurrency group) or use an external lock. Runners for different namespaces may share the history table; a history write that loses a race against other namespaces’ writes is retried, up to the history table’s commit.retry.num-retries.

Versions¶

Migration files are named NNNN_<description>.py and the Migration.version attribute must equal the NNNN prefix. A target argument accepts 1-4 digits ("2", "0002") or an int and is normalised with normalize_version(), so "2" and "10" order numerically.

Usage¶

from iceberg_ops.core.catalog import CatalogFactory
from iceberg_ops.runner       import MigrationRunner

catalog = CatalogFactory.build("sqlite")

runner = MigrationRunner(
    catalog,
    namespace="sales",                     # required: target namespace
    meta_namespace="_iceberg_ops",         # optional: where history lives
    migrations_dir="./migrations",         # optional: migration files
)

runner.status()                            # show pending / applied state
runner.migrate()                           # apply all pending migrations
runner.migrate(target="0002")              # apply up to version 0002
                                           # ("2" and 2 work too)
runner.rollback(steps=1)                   # undo the last migration
runner.rollback(target="0001")             # roll back until 0001 is latest
runner.stamp()                             # mark migrations applied
                                           # without running upgrade(),
                                           # used to adopt iceberg-ops on
                                           # an existing schema.
runner.history()                           # print each version's latest
                                           # recorded status

iceberg_ops.runner.normalize_version(value)[source]¶

Normalise a user-supplied migration version to its 4-digit form.

Parameters:: value (str | int) – A version such as "2", "0002" or 2. Surrounding whitespace is ignored.
Returns:: The zero-padded 4-digit version, e.g. "0002".
Return type:: str
Raises:: ValueError – If value is not 1-4 decimal digits.

Example

>>> normalize_version("2")
'0002'
>>> normalize_version(10)
'0010'

class iceberg_ops.runner.MigrationRunner(catalog, namespace, meta_namespace='_iceberg_ops', migrations_dir='./migrations')[source]¶

Bases: object

Orchestrates discovery, application, and rollback of Iceberg migrations for a single target namespace.

Parameters:

catalog (Any) – A live PyIceberg catalog (from iceberg_ops.core.catalog.CatalogFactory.build() or BYOC).
namespace (str) – Required. The namespace whose tables this runner manages (e.g. "sales"). History is scoped to this namespace so runners for different namespaces can share the same catalog and history table. Runs for the same namespace are not guarded against each other: run one job per namespace at a time.
meta_namespace (str, default: '_iceberg_ops') – Namespace used internally to store migration history. Default: "_iceberg_ops". Should not contain user tables.
migrations_dir (str, default: './migrations') – Directory containing NNNN_<description>.py migration files. Default: "./migrations" (relative to the current working directory).

status()[source]¶

Print the status of every discovered migration for this namespace.

Legend: [applied] [pending] [rolled_back] [FAILED]. History rows whose migration file is no longer discovered are listed too, marked (migration file missing). Does not write to the catalog (the history table is not created), but imports every migration file, which runs its module-level code.

Raises:

FileNotFoundError – If the migrations directory does not exist.
RuntimeError – If a migration file cannot be loaded.

Return type:

None

history()[source]¶

Print all history records for this namespace.

Return type:: None

plan(target=None)[source]¶

Preview pending migrations without applying them.

No tables are created (not even the history table) and no history records are written.

Parameters:

target (str | int | None, default: None) – Show only pending migrations whose version is at most this one (inclusive). Accepts 1-4 digits ("2", "0002") or an int; it must name a discovered migration.

Raises:

ValueError – If target is malformed or names no discovered migration.
FileNotFoundError – If the migrations directory does not exist.
RuntimeError – If a migration file cannot be loaded.

Return type:

None

migrate(target=None)[source]¶

Apply all pending migrations up to target (or all if target is None).

Migrations are applied in ascending version order. Any failure is recorded as "failed" and the runner stops immediately.

Parameters:

target (str | int | None, default: None) – Apply only up to (and including) this version. Accepts 1-4 digits ("2", "0002") or an int; it must name a discovered migration.

Return type:

None

Raises:

ValueError – If target is malformed or names no discovered migration. Nothing is applied.
FileNotFoundError – If the migrations directory does not exist.
RuntimeError – If a migration file cannot be loaded, or a migration fails (the failure is recorded as "failed"; if that history write itself fails, it is logged and the migration’s own error is still the one raised).
ConcurrentMigrationError – If a migration’s own commit or its history write conflicted with a concurrent commit, nothing is recorded for that version. If another runner’s versions appear in history after this run recorded a version, this run’s version stays recorded as applied.

Notes

Once a version is recorded as applied, later errors (from the history re-read that looks for other runners, for example) propagate unchanged and never turn that row into "failed". A failure is recorded as "<ExceptionType>: <message>"; the traceback is only logged locally, at DEBUG level.

rollback(steps=1, target=None)[source]¶

Roll back applied migrations (highest version first).

Like migrate(), concurrent runs for one namespace are not guarded: run one job per namespace at a time.

Parameters:

steps (int, default: 1) – Number of migrations to undo; must be at least 1. Ignored when target is set.
target (str | int | None, default: None) – Roll back until the last applied version is target (i.e. undo all versions strictly greater than target). Accepts 1-4 digits or an int; "0000" undoes every applied migration. Must name a discovered or applied version.

Return type:

None

Raises:

ValueError – If steps is below 1 (without target), or target is malformed or unknown. Nothing is rolled back.
FileNotFoundError – If the migrations directory does not exist.
RuntimeError – If a migration file to undo is missing (checked before any downgrade() runs, so nothing is rolled back), a file cannot be loaded, or a downgrade() fails.
NotImplementedError – If a migration’s downgrade() is not implemented.
ConcurrentMigrationError – If another runner recorded history for this namespace during the run, or a migration’s own commit conflicted with a concurrent commit.

Notes

Once a version is recorded as rolled back, later errors (from the history re-read that looks for other runners, for example) propagate unchanged rather than as a Rollback of ... failed error.

stamp(target=None)[source]¶

Mark migrations up to target as applied without running upgrade().

Used to adopt iceberg-ops on a database whose schema was created out-of-band (e.g. an existing Iceberg deployment that predates the framework). For each migration whose current status is not applied (never run, failed or rolled_back), an applied row replaces any existing row for that version. A failed version is therefore marked applied without being re-run, even if its upgrade() stopped halfway: check such versions before stamping. No pre_upgrade / upgrade / post_upgrade hook runs and no user tables are created. Idempotent — re-running is a no-op.

Parameters:

target (str | int | None, default: None) – Latest version to stamp (inclusive). None (default) stamps every discovered migration. Accepts 1-4 digits ("2", "0002") or an int; it must name a discovered migration.

Raises:

ValueError – If target is malformed or names no discovered migration.
FileNotFoundError – If the migrations directory does not exist.
RuntimeError – If a migration file cannot be loaded.
ConcurrentMigrationError – If another runner recorded history for the same version concurrently.

Return type:

None

Notes

stamp does not validate that the user’s actual schema matches the stamped baseline. It simply updates the history. If your schema diverges from what migrations 0001…N would have produced, a subsequent migrate may still fail when migration N+1 runs against a schema it did not expect.