alembic-static-data

An Alembic plugin that loads static reference data (currencies, countries, lookup tables, ...) from a CSV file in a migration, idempotently, keyed on natural keys. Alembic has no built-in support for repeatable/seed data; this fills that gap.

Status: first prototype (0.1.0a1). Not for production use.

Pre-1.0 the static_data_version table's own shape may change between releases, and it is deliberately not migrated. If a release adds a column, a table written by an earlier one is refused with an error that says so; drop it and it is recreated on the next load. It holds bookkeeping only — never your data. After 1.0 that stops being acceptable and any change there will be handled properly.

Usage

# in a migration script
from alembic import op


def upgrade() -> None:
    op.load_static_data("currency", "data/currency.csv", natural_keys=["code"])

currency.csv:

code,name,decimals
EUR,Euro,2
JPY,Yen,0

op.load_static_data(table_name, csv_path, natural_keys, schema=None, *, delete_missing=False, track_checksum=True, version_table_name="static_data_version", environment_column=None, environment_key="env", require_unique=True):

• The natural keys must identify at most one row, and the table has to prove it. The load refuses unless a primary key, UniqueConstraint or unique Index is covered by the natural keys, because the upsert's UPDATE ... WHERE <keys> would otherwise overwrite every matching row while reporting one. A unique key only has to be a subset of the natural keys: uniqueness on code justifies natural_keys=["code", "name"], but a unique constraint on (code, name) does not justify natural_keys=["code"]. require_unique=False opts out, for schemas where uniqueness is enforced somewhere this cannot see.
• A single natural key may be given as a plain string: natural_keys="code".
• Rows whose natural key is not in the table are inserted.
• Rows whose natural key exists and whose non-key values differ are updated (only the changed columns).
• Identical rows are left alone, so running it again changes nothing (idempotent).
• Rows in the table but absent from the CSV are never deleted unless you pass delete_missing=True. Columns absent from the CSV are never touched either way.
• The table must already exist (create it earlier in the migration). It is reflected from the live connection.
• Cells are converted to the column's Python type (int, float, Decimal, bool, date, datetime ISO format); an empty cell becomes NULL.
• The returned UpsertResult(inserted, updated, unchanged, deleted, skipped) is also logged at INFO.
• Uses plain SQLAlchemy Core (SELECT, then INSERT or UPDATE), no dialect-specific ON CONFLICT/MERGE.
• csv_path is handed straight to open(), so a relative path resolves against the working directory, not the migration file. Build an absolute path from the revision's own __file__ instead — see the worked example; env.py cannot hold that constant, because Alembic loads it by path rather than as an importable module.

Failures raise subclasses of alembic_static_data.errors.StaticDataError with a message naming the file, line, column or key. The CSV itself: a missing or empty file, a blank or duplicated header column, a row with the wrong number of cells, a natural key missing from the header, a duplicate natural key, a column the table does not have, a value that cannot be converted. The call: natural keys the table does not guarantee to be unique, environment_column without an environment to select with, and delete_missing=True combined with environment_column. The context: offline (--sql) mode, which cannot work because the operation must read the database, and a static_data_version table left over from an earlier pre-1.0 release.

Environment scoping

One CSV can carry rows for several environments. Add a column naming them, and pass the environment on the command line:

code,name,decimals,_env
EUR,Euro,2,
JPY,Yen,0,
XTS,Test currency,2,dev
XXX,Sandbox,2,dev;staging

op.load_static_data("currency", CSV_PATH, natural_keys=["code"], environment_column="_env")

alembic -x env=prod upgrade head     # loads EUR, JPY
alembic -x env=dev upgrade head      # loads EUR, JPY, XTS, XXX

• A blank cell means the row applies everywhere; otherwise the cell is a ;-separated list (, belongs to the CSV), compared case-insensitively after stripping.
• The column is not a table column: it is removed before the upsert, and it may not be a natural key.
• The same natural key may appear once per environment, which is how you override a row for dev. Two rows for the selected environment are still a duplicate-key error.
• The environment comes from Alembic's own -x mechanism, read with get_x_argument(). The key defaults to env and is configurable via environment_key=.
• Rows for other environments are ignored, not deleted — scoping selects what to load, it does not reconcile what is already there. delete_missing=True is therefore refused together with environment_column: the filtered CSV omits every other environment's rows, the table has no environment column to tell them apart, and deleting what looks missing would wipe them. Give each environment its own database if you need the CSV to be authoritative.
• Configuring environment_column= with no environment available is an error (EnvironmentNotSetError), not a reason to quietly load a subset.

The environment is part of the checksum key, so loading the same file for a second environment is not skipped, and op.forget_static_data(..., environment="dev") forgets just that one.

Worked example

examples/reference_data is a complete, runnable Alembic project using this plugin — alembic.ini, env.py, two revisions and a CSV. tests/test_example_project.py drives it through alembic.command, so it is verified rather than decorative, and it is the only place the plugin is exercised the way a user meets it: entry-point discovery inside a real env.py, a real --autogenerate run, and alembic downgrade base && alembic upgrade head putting the data back.

Rerun tracking

By default each load records its CSV's SHA-256 in a static_data_version table (created on first use, in the same schema as the table being loaded, next to Alembic's alembic_version), and skips the load entirely while that checksum is unchanged:

op.load_static_data("currency", "data/currency.csv", natural_keys=["code"])
# first run:  UpsertResult(inserted=2)
# re-run:     UpsertResult(skipped=True)      - file unchanged, nothing read, nothing written
# after edit: UpsertResult(updated=1)         - checksum differs, so the upsert runs again

This is what makes a revision safe to re-run when its data file changes, which is Alembic's missing repeatable-migration behaviour (issue #1). Notes on the design:

• The trigger lives inside the operation, not in an env.py hook or a separate command, so data loads stay ordered by revision order and nothing new has to be wired up.
• The checksum describes the file, not the table. A row edited by hand in the database stays edited until the CSV itself changes; track_checksum=False restores the old always-upsert behaviour.
• Bytes are hashed, not parsed rows, so re-indenting a CSV counts as a change (the following load is then a no-op upsert). Line endings count too: the same file as CRLF and as LF hash differently.
• One row per (table_name, csv_path, environment), keyed on the path as written in the migration with separators normalised - so a relative path stays stable across checkouts, the same CSV loaded into two tables is tracked separately, and each environment is tracked on its own (see Environment scoping above).
• version_table_name= renames the table; loads tracked in different tables do not see each other.

Downgrades

The recorded checksum outlives the data, so a downgrade() that removes the rows must also forget the load, or the next upgrade sees a matching checksum and skips the reload:

def upgrade() -> None:
    op.load_static_data("currency", "data/currency.csv", natural_keys=["code"])


def downgrade() -> None:
    op.forget_static_data("currency", "data/currency.csv")

op.forget_static_data(table_name, csv_path=None, schema=None, *, version_table_name="static_data_version", environment=None) deletes only bookkeeping, never table data, and returns how many recorded loads it forgot. Omit csv_path to forget every file recorded for that table, and environment to forget every environment. It never creates the checksum table, and forgetting something that was never recorded returns 0 rather than failing.

Autogenerate

The checksum table exists in the database but not in your target_metadata, so alembic revision --autogenerate would otherwise propose dropping it. Exclude it in env.py:

from alembic_static_data import exclude_from_autogenerate

context.configure(
    connection=connection,
    target_metadata=target_metadata,
    include_name=exclude_from_autogenerate,
)

For a renamed table, bind the name first: functools.partial(exclude_from_autogenerate, version_table_names=["my_static_data"]). If you already have an include_name, call both and and the results.

Install

Requires Python 3.11+ and Alembic 1.17 or newer.

pip install git+https://github.com/rt87/alembic-static-data   # repo is private for now
# or, from a checkout
uv pip install -e .

On Alembic 1.18 and newer Alembic discovers the plugin through the alembic.plugins entry point automatically; no env.py change is needed, and importing Alembic registers op.load_static_data.

On Alembic 1.17, which has no plugin system, register the operation from env.py:

from alembic_static_data import register_operations

register_operations()

That call is idempotent and harmless on every version, so it is safe to keep in an env.py that has to run against a range of Alembic releases.

Releasing

.github/workflows/release.yml builds, checks and publishes — manually only, from the Actions tab, with the index as an input (testpypi or pypi). The tag trigger is left commented out so a stray tag can never publish, and so it costs no CI minutes until a release is actually wanted.

It uses Trusted Publishing: no API token exists anywhere in this repository or in its secrets. The publish job authenticates with a short-lived OIDC identity, which is why it is the only job granted id-token: write — the build job, which runs project code, cannot mint one.

Three things must match the pending publisher configured on the index, or the upload is rejected:

	value
Repository	rt87/alembic-static-data
Workflow filename	release.yml
Environment	(none) — add environment: to the publish job if one was configured

TestPyPI and PyPI are separate configurations; a pending publisher on one does not authorise the other. Before the first upload the build job installs the freshly built wheel into a clean environment and asserts that Alembic discovers the plugin and that py.typed is packaged — the one check the test suite cannot do, since it runs against the editable install.

Development

git config core.hooksPath .githooks   # once per clone, see below
uv sync
uv run ruff check .
uv run black .       # formats at line-length 120; `blackd` is installed for editors
uv run mypy          # strict, over src/ and tests/
uv run vermin -t=3.11- --violations --no-tips src tests   # does the code really still fit 3.11?
uv run pytest        # includes coverage; pytest-randomly shuffles order each run
uv run pytest --testmon --no-cov   # fast inner loop: only tests affected by your edits
uv run pytest -n auto --no-cov     # parallel

Identity hooks

.githooks/ holds two hooks that keep commits from carrying anything that should not become public: commit-msg fails a commit immediately, and pre-push inspects every commit about to leave the machine - its author and committer identity, its message, and the content it introduces. Enable them once per clone with git config core.hooksPath .githooks.

The allow-list is this account's GitHub noreply address plus GitHub's own address on web merge commits; any other address fails. Project-wide patterns live in .githooks/identity-check and are generic. Anything personal - a corporate login, an employer name, a private address - goes in .githooks/identity-check.local, which is gitignored: a committed rule would publish the very strings it exists to withhold. Copy identity-check.local.example to get started.

Both are a seatbelt, not a lock: git push --no-verify bypasses them, and a clone that never set core.hooksPath has no hooks at all.

CI lives in .github/workflows/ci.yml and runs on every push to main and every pull request against it; public repositories get Actions minutes for free. It lints, type-checks, checks the Python floor with vermin, and runs the suite on Python 3.11-3.14 plus one job per Alembic plugin-loading regime. The four full-suite jobs enforce --cov-fail-under=100, so the coverage badge cannot quietly start lying; the older-Alembic jobs skip the gate because two plugin-system tests skip themselves below 1.18, which makes the percentage incomparable.

The suite covers in-memory SQLite and, via testcontainers, a real Postgres 17 - the latter is the only way to exercise named schemas and server-side Numeric. The Postgres tests skip themselves when Docker is not reachable, so uv run pytest works either way.

To check an older Alembic (the suite skips the two plugin-system tests below 1.18):

uv run --with "alembic==1.17.0" pytest -q --no-cov -W error::DeprecationWarning

Not done yet

• The load has no inverse for the data it wrote - only for its bookkeeping - so a downgrade() that removes rows has to call op.forget_static_data() itself; see Downgrades above.
• Deleting rows is opt-in per call (delete_missing=True) and is skipped along with everything else when the checksum matches, so it is not a reconciliation mechanism.
• No offline (--sql) support.
• Rows are processed one at a time (one SELECT each), which is fine for reference data but not for large files.
• With require_unique=False, updated counts CSV rows rather than affected table rows, so it under-reports when one CSV row matches several table rows. That is the whole reason the check is on by default.

Alembic compatibility

Alembic changed plugin loading twice in the past year, so the entry point has to cope with three regimes:

Alembic	Released	Plugin system	How the operation gets registered
1.17.x	2025-10	none	register_operations() in env.py
1.18.0 – 1.20.0	2026-01 – 2026-09	yes, loader iterates entrypoint.load()	entry point, automatically
1.20.1+	2026-10	yes, loader expects one module with setup()	entry point, automatically

1.18.0 was the first release with a plugin system, and through 1.20.0 its loader ran for mod in entrypoint.load(), contradicting the docs, which say to point the entry point at a single module with setup(). A plain module therefore raises TypeError while import alembic is running. 1.20.1 fixed the loader (Alembic #1873) to accept a module, and accepts the old collection form only with a deprecation warning.

alembic_static_data.plugin satisfies both: it is a module with setup() whose type also iterates as a one-element collection containing the module itself (_PluginModule). New Alembic takes the module branch, older Alembic iterates and arrives at the same module, and neither path emits a warning.

License

MIT, matching Alembic and SQLAlchemy, so this code can move upstream without a relicensing step.