Skip to content
AITroveRead. Build. Understand.
Reading list
Make this comfortable

Python SQLite migration: reopen the file after a failed upgrade

Last updated: 1 Oct 20265 min read
tutorial
IntermediateBy AITrove Editorial

A migration rollback should be checked through a fresh connection to the same database file.

Download Python source kit

Operation contract

Version one is committed to a temporary disk database. A second connection attempts a column change and then fails on a duplicate table. After rollback and close, a third connection reads version and columns from the file. Reopening catches a class of mistakes that an in-memory connection-only assertion cannot show.

Failure and ownership boundary

A temporary file is still not a production rehearsal. The fixture does not check lock contention, a large table rewrite, power loss, multiple app versions, or restoration from backup. Run the exact release sequence against a copy of realistic data and hold one migration owner at a time.

Working program

python
import sqlite3
import tempfile
from contextlib import closing
from pathlib import Path

with tempfile.TemporaryDirectory() as directory:
    database_path = Path(directory) / "receipts.db"
    with closing(sqlite3.connect(database_path, isolation_level=None)) as database:
        database.execute("BEGIN IMMEDIATE")
        database.execute("CREATE TABLE receipts (receipt_id TEXT PRIMARY KEY)")
        database.execute("PRAGMA user_version = 1")
        database.execute("COMMIT")

    with closing(sqlite3.connect(database_path, isolation_level=None)) as database:
        try:
            database.execute("BEGIN IMMEDIATE")
            database.execute("ALTER TABLE receipts ADD COLUMN batch_id TEXT")
            database.execute("CREATE TABLE receipts (duplicate_id TEXT)")
            database.execute("PRAGMA user_version = 2")
            database.execute("COMMIT")
        except sqlite3.OperationalError:
            database.execute("ROLLBACK")

    with closing(sqlite3.connect(database_path)) as database:
        version = database.execute("PRAGMA user_version").fetchone()[0]
        columns = [row[1] for row in database.execute("PRAGMA table_info(receipts)")]
    print("version", version)
    print("columns", columns)

Output

Output
version 1
columns ['receipt_id']

Costs and limits

A fresh connection adds file I/O and catches committed-state mistakes. Schema rewrites can block writers and need measurement at real data size.

Common Mistakes

  • Do not infer file state from one still-open connection alone.
  • Updating the version marker before the schema succeeds mislabels the database.
  • This fixture does not simulate a killed process mid-upgrade.

Connected lessons

Test this boundary.

python
sqlite-migration-reopen-check
Storage details