Adding a Required Foreign Key to a Table That Already Has Rows: A Safe Alembic Migration

The problem

A developer in an online forum had a working app with two tables, users and images, both full of data. They wanted every image to belong to a user. So they added a foreign key from images to users, marked it "not null", and generated a migration with Flask-Migrate (which uses Alembic).

Before running flask db upgrade, they asked a fair question. What happens to the rows that are already there? Will the upgrade delete them? Will the new rule just skip the old rows? And how do you change a database that people are already using?

This post answers all three, with code we ran on SQLite.

The root cause

A migration changes the shape of the database. It does not invent data. A new column on a full table starts out empty (NULL) in every old row. A "not null" rule says the column can never be empty. Both cannot be true at once, so the database refuses the change.

Alembic's autogenerate cannot fix this for you. It compares your models with the database and writes the shape change. It does not know which user should own each old image. Only you know that.

See it fail

We built a small copy of the setup: Alembic with SQLAlchemy models, a SQLite file, 2 users and 3 images. Then we added this line to the Image model:

user_id: Mapped[int] = mapped_column(ForeignKey("users.id")) # new: NOT NULL

With Mapped[int] (not Mapped[int | None]), SQLAlchemy treats the column as not null. Autogenerate wrote this:

alembic revision --autogenerate -m "link images to users" --rev-id 0002

def upgrade() -> None: with op.batch_alter_table('images', schema=None) as batch_op: batch_op.add_column(sa.Column('user_id', sa.Integer(), nullable=False)) batch_op.create_foreign_key(batch_op.f('fk_images_user_id_users'), 'users', ['user_id'], ['id'])

Running it on the table with 3 images:

alembic upgrade head

sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) NOT NULL constraint failed: _alembic_tmp_images.user_id [SQL: INSERT INTO _alembic_tmp_images (id, img_link) SELECT images.id, images.img_link FROM images]

So, does it delete your data?

No. In our test, all 3 images were still there after the failure, and alembic current still showed the old revision, 0001.

But look at the table name in the error: _alembic_tmp_images. SQLite can only do a few kinds of ALTER TABLE. So for changes like this, Alembic uses "batch mode": it makes a new table, copies the rows into it, drops the old table and renames the new one. The copy step is what failed. Alembic told us it would "assume non-transactional DDL" on SQLite, and the half-built temp table was left behind:

[('alembic_version',), ('sqlite_autoindex_alembic_version_1',), ('images',), ('users',), ('_alembic_tmp_images',)]

If you fix the migration and run it again without cleaning up, you get a second, more confusing error:

sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) table _alembic_tmp_images already exists

Drop the leftover table first (after you make a backup):

import sqlite3 with sqlite3.connect("app.db") as db: db.execute("DROP TABLE IF EXISTS _alembic_tmp_images")

The fix: expand, backfill, contract

Edit the generated migration so it does three small steps in order:

Expand. Add the column, but allow empty values for now.
Backfill. Give every old row a value.
Contract. Now add the "not null" rule and the foreign key.

The hard part is step 2, because you must decide who owns the old images. In this demo, they go to one placeholder account. The .invalid ending is reserved, so that address can never reach a real inbox. In a real app, you might match images to users by another column, or delete images nobody owns. That is a product decision, not a database one.

# migrations/versions/0002_link_images_to_users.py from alembic import op import sqlalchemy as sa revision = "0002" down_revision = "0001" branch_labels = None depends_on = None PLACEHOLDER_EMAIL = "unassigned@example.invalid" # Small table stand-ins, so this file never imports the app's models. users = sa.table("users", sa.column("id", sa.Integer), sa.column("email", sa.String)) images = sa.table("images", sa.column("user_id", sa.Integer)) def upgrade() -> None: # Step 1 (expand): add the column, but allow empty values for now. with op.batch_alter_table("images") as batch_op: batch_op.add_column(sa.Column("user_id", sa.Integer(), nullable=True)) # Step 2 (backfill): give every old image an owner. conn = op.get_bind() op.execute(users.insert().values(email=PLACEHOLDER_EMAIL)) owner_id = conn.execute( sa.select(users.c.id).where(users.c.email == PLACEHOLDER_EMAIL) ).scalar_one() op.execute( images.update().where(images.c.user_id.is_(None)).values(user_id=owner_id) ) # Step 3 (contract): now that no row is empty, add the rule and the key. with op.batch_alter_table("images") as batch_op: batch_op.alter_column("user_id", existing_type=sa.Integer(), nullable=False) batch_op.create_foreign_key( batch_op.f("fk_images_user_id_users"), "users", ["user_id"], ["id"] ) def downgrade() -> None: with op.batch_alter_table("images") as batch_op: batch_op.drop_constraint( batch_op.f("fk_images_user_id_users"), type_="foreignkey" ) batch_op.drop_column("user_id") op.execute(users.delete().where(users.c.email == PLACEHOLDER_EMAIL))

Why sa.table() and not the app's Image model? Models change over time. A migration that imports them can break a year later, when the model no longer matches this point in history. The small stand-ins only describe the columns this one step needs.

Two setup lines that matter on SQLite

In migrations/env.py, turn on batch mode for autogenerate, so new migrations use the copy-and-swap method:

context.configure( connection=connection, target_metadata=target_metadata, render_as_batch=True, # SQLite cannot ALTER columns in place )

And give your constraints names. Batch mode on SQLite needs a name to find a foreign key again, for example when you drop it in downgrade(). A naming convention on the metadata does this for every table:

NAMING = { "pk": "pk_%(table_name)s", "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s", "uq": "uq_%(table_name)s_%(column_0_name)s", "ix": "ix_%(column_0_label)s", "ck": "ck_%(table_name)s_%(constraint_name)s", } class Base(DeclarativeBase): metadata = MetaData(naming_convention=NAMING)

Run it, undo it, run it again

We started from a fresh database with 2 users and 3 images, then ran the fixed migration. A small show.py script prints the tables and the column rules.

alembic upgrade head python show.py

INFO [alembic.runtime.migration] Running upgrade 0001 -> 0002, link images to users users: [(1, 'a@example.com'), (2, 'b@example.com'), (3, 'unassigned@example.invalid')] images: [(1, '/img/1.png', 3), (2, '/img/2.png', 3), (3, '/img/3.png', 3)] user_id NOT NULL: True foreign keys: [('users', 'user_id', 'id')] leftover temp tables: []

Every old image kept its link and got an owner. The rule and the key are in place.

alembic downgrade -1 python show.py

INFO [alembic.runtime.migration] Running downgrade 0002 -> 0001, link images to users users: [(1, 'a@example.com'), (2, 'b@example.com')] images: [(1, '/img/1.png'), (2, '/img/2.png'), (3, '/img/3.png')] foreign keys: [] leftover temp tables: []

Running alembic upgrade head again gave the same result as the first time. Then alembic check confirmed the models and the database agree:

No new upgrade operations detected.

Test it so it stays fixed

Migrations are code, so test them. Each test gets its own empty database from pytest's tmp_path.

import sqlite3 import pytest from alembic import command from alembic.config import Config def make_config(db_path): cfg = Config("alembic.ini") cfg.set_main_option("sqlalchemy.url", f"sqlite:///{db_path}") return cfg def test_up_down_up(tmp_path): cfg = make_config(tmp_path / "test.db") command.upgrade(cfg, "head") command.downgrade(cfg, "base") command.upgrade(cfg, "head") def test_old_images_survive_and_get_an_owner(tmp_path): db_path = tmp_path / "test.db" cfg = make_config(db_path) command.upgrade(cfg, "0001") with sqlite3.connect(db_path) as db: db.execute("INSERT INTO images (img_link) VALUES ('/img/1.png')") command.upgrade(cfg, "head") with sqlite3.connect(db_path) as db: rows = db.execute( "SELECT i.img_link, u.email FROM images i JOIN users u ON u.id = i.user_id" ).fetchall() assert rows == [("/img/1.png", "unassigned@example.invalid")] def test_owner_is_required_at_head(tmp_path): db_path = tmp_path / "test.db" command.upgrade(make_config(db_path), "head") with sqlite3.connect(db_path) as db: with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"): db.execute("INSERT INTO images (img_link) VALUES ('/img/9.png')") def test_owner_must_exist_when_fks_are_on(tmp_path): db_path = tmp_path / "test.db" command.upgrade(make_config(db_path), "head") with sqlite3.connect(db_path) as db: db.execute("PRAGMA foreign_keys = ON") with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"): db.execute("INSERT INTO images (img_link, user_id) VALUES ('/x.png', 999)")

4 passed in 0.34s

Note the PRAGMA foreign_keys = ON line. SQLite only checks foreign keys when this is turned on for the connection.

Trade-offs

A placeholder owner is honest but visible. Your app must know that "unassigned" images exist. If no image should exist without a real owner, delete or fix those rows in the backfill instead.
server_default is the shortcut. If one fixed value fits every old row, you can add the column with server_default and skip the UPDATE. A plain default= on the model does not help here: that default is filled in by SQLAlchemy when your app inserts a row, not by the database for rows that already exist.
This migration cannot run in offline mode. It reads the placeholder's id from the live database. When we tried alembic upgrade 0001:0002 --sql, it failed with 'NoneType' object has no attribute 'scalar_one'. If your team reviews SQL before release, use a fixed id or a subquery instead.
Batch mode copies the whole table. That is fine for small SQLite tables. On a big, busy production database, the UPDATE and the copy can take a long time and lock the table. Back up first, try it on a copy of real data, and think about filling rows in smaller batches.
One file or three? We kept all three steps in one migration because the app is offline during the upgrade. If your app keeps running while you deploy, split them into separate migrations and ship code in between.

How to check it worked

Back up the database file before you run anything.
Run alembic upgrade head, then alembic current. It should show the new revision.
Count rows before and after. The count should not change.
Look for rows the backfill missed: SELECT COUNT(*) FROM images WHERE user_id IS NULL; should return 0.
Look for leftovers: SELECT name FROM sqlite_master WHERE name LIKE '_alembic_tmp_%'; should return nothing.
Run alembic check. It should say "No new upgrade operations detected."
Run the up-down-up test in CI on every change.