Complete worked example
python
import sqlite3
from contextlib import closing
from pathlib import Path
from tempfile import TemporaryDirectory
from sqlalchemy import create_engine, text, inspect
from alembic import command
from alembic.config import Config
from alembic.migration import MigrationContext
ENV_SOURCE = """\
from alembic import context
connection = context.config.attributes["connection"]
context.configure(connection=connection, target_metadata=None)
with context.begin_transaction():
context.run_migrations()
"""
BASE_SOURCE = """\
from alembic import op
import sqlalchemy as sa
revision = "owned_base"
down_revision = None
def upgrade():
op.create_table("roles", sa.Column("id", sa.Integer, primary_key=True),
sa.Column("score", sa.Integer, nullable=False))
def downgrade():
op.drop_table("roles")
"""
LABEL_SOURCE = """\
from alembic import op
import sqlalchemy as sa
revision = "owned_label"
down_revision = "owned_base"
def upgrade():
op.add_column("roles", sa.Column("label", sa.String(30), nullable=True))
def downgrade():
with op.batch_alter_table("roles") as batch:
batch.drop_column("label")
"""
def configure_owned(folder):
scripts = Path(folder) / "migrations"
(scripts / "versions").mkdir(parents=True)
(scripts / "env.py").write_text(ENV_SOURCE, encoding="utf-8")
(scripts / "versions" / "owned_base.py").write_text(BASE_SOURCE, encoding="utf-8")
(scripts / "versions" / "owned_label.py").write_text(LABEL_SOURCE, encoding="utf-8")
config = Config()
config.set_main_option("script_location", str(scripts))
return config
def copy_owned_database(source_path, target_path):
with closing(sqlite3.connect(source_path, autocommit=True)) as source:
with closing(sqlite3.connect(target_path, autocommit=True)) as target:
source.backup(target)
def inspect_example(scores, increment):
with TemporaryDirectory() as folder:
database, backup = Path(folder) / "owned.sqlite3", Path(folder) / "backup.sqlite3"
config = configure_owned(folder)
engine = create_engine("sqlite+pysqlite:///" + str(database), connect_args={"autocommit": False})
try:
with engine.begin() as connection:
config.attributes["connection"] = connection
command.upgrade(config, "owned_base")
if scores:
connection.execute(text("INSERT INTO roles (id, score) VALUES (:id, :score)"),
[{"id": i, "score": s} for i, s in enumerate(scores, 1)])
engine.dispose() # no application writers remain in this owned fixture
copy_owned_database(database, backup)
with engine.begin() as connection:
config.attributes["connection"] = connection
command.upgrade(config, "head")
connection.execute(text("UPDATE roles SET score = score + :increment"), {"increment": increment})
changed = list(connection.scalars(text("SELECT score FROM roles ORDER BY id")))
upgraded = MigrationContext.configure(connection).get_current_revision()
engine.dispose()
copy_owned_database(backup, database)
with engine.connect() as connection:
restored = list(connection.scalars(text("SELECT score FROM roles ORDER BY id")))
revision = MigrationContext.configure(connection).get_current_revision()
has_label = any(c["name"] == "label" for c in inspect(connection).get_columns("roles"))
return {"upgraded": upgraded, "changedScores": changed, "restoredRevision": revision,
"restoredScores": restored, "restoredHasLabel": has_label}
finally:
engine.dispose()- Reasoning
- After the backup, the program upgrades to owned_label and changes the stored scores by a bound increment. It records that migrated state, disposes its own Engine pool again, restores the backup into the owned file and opens a fresh connection. For scores [3, 7] and increment 5, changedScores is [8, 12], but restoredScores returns to [3, 7]. The restored revision is owned_base and the restored schema no longer has label. Inspecting all three observations makes the recovery concrete: an earlier version string alone would not establish that rows and columns were restored.