Typorithm · hard

Capstone: migrate, inspect and restore an owned database

pythonsqlalchemyalembicdatabasesindependent-practice
Saved on this device
Full lesson · 4 learning steps Open any exercise for focused practice
1

Step 1 of 4 · intro card

Capstone: migrate, inspect and restore an owned database

0%

Concept and contract

Verify both revision history and data after a real SQLite backup and restoration.

Recovery needs evidence about a complete earlier database state, not only a successful command or a remembered revision name. A migration downgrade runs reviewed Python operations; restoring a backup reinstates the state captured by that backup. These actions can have different schema and data outcomes.

The capstone owns a temporary SQLite file and two Alembic revisions. It commits base rows, closes its own pooled connections and uses SQLite’s backup API to create an independently reopenable earlier image. No application outside this private fixture writes to the database, which is a crucial boundary for the experiment.

Mechanism and worked example

After the backup, the program upgrades to owned_label and changes the stored scores by a bound increment.

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.

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.
Download the SQLAlchemy and Alembic computer lab ↗

Focused decision: context is in the complete example

python

engine.dispose()
copy_owned_database(backup, database)
with engine.connect() as connection:
    revision = MigrationContext.configure(connection).get_current_revision()
Database decision
The backup and restore use actual SQLite copy operations with closed owned handles. Reopening verifies data and schema together; the task adds a total computed from that restored database.
Download the SQLAlchemy and Alembic computer lab ↗

Counterexamples and common mistakes

The controlled mistake skips the actual restoration call.

The controlled mistake skips the actual restoration call. The program still reopens the file, but it sees the migrated head, changed scores and added label column. A message saying restore succeeded would therefore be contradicted by the actual database state.

Engine.dispose does not stop other processes, cancel every checked-out transaction or quiesce a real service. This fixture has no such writers by construction. Likewise, restoring the earlier image discards the fixture’s later score updates; a production recovery plan must account for accepted writes after the backup rather than pretending that any older image preserves them automatically.

Controlled bug: observe the rejected database contract

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()
            # Missing the actual restoration
            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
The controlled mistake skips the actual restoration call. The program still reopens the file, but it sees the migrated head, changed scores and added label column. A message saying restore succeeded would therefore be contradicted by the actual database state. Engine.dispose does not stop other processes, cancel every checked-out transaction or quiesce a real service. This fixture has no such writers by construction. Likewise, restoring the earlier image discards the fixture’s later score updates; a production recovery plan must account for accepted writes after the backup rather than pretending that any older image preserves them automatically.
Download the SQLAlchemy and Alembic computer lab ↗

Independent practice

First type the disposal, backup restoration and fresh revision check.

First type the disposal, backup restoration and fresh revision check. For independent practice read COALESCE(SUM(score), 0) from the restored database and include restoredTotal in the report. Compute it through the reopened connection rather than summing the original Python argument beside an unverified restore.

Keep the earlier migrated-state observations as well as the restored rows, revision and column checks. Use positive values, a zero with a negative value, and an empty table. The empty table is important because SQL SUM alone returns NULL; the requested report contract instead requires an explicit zero.

Step 1: Read a COALESCE(SUM(score), 0) total from the restored database and include restoredTotal in the report. Step 2: Read the total through the fresh restored connection and retain every existing recovery observation.

Practice starter: solve before reading the solution

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()

# Task: Read a COALESCE(SUM(score), 0) total from the restored database and include restoredTotal in the report.
Reasoning
Read a COALESCE(SUM(score), 0) total from the restored database and include restoredTotal in the report.
Download the SQLAlchemy and Alembic computer lab ↗
  • Hint 1: Add a scalar aggregate query beside the restored row and revision reads.
  • Hint 2: Use coalesce(sum(score), 0) and test the empty fixture.

Solution, tests and trade-offs

The solution queries the restored database for its total and returns ten for [3, 7], minus two for [0, -2], and zero for no rows.

The solution queries the restored database for its total and returns ten for [3, 7], minus two for [0, -2], and zero for no rows. The rest of the recovery report remains unchanged. An omitted restore cannot satisfy the complete observed state merely by copying a total from the input.

All database handles in the copy helper are explicitly closed; sqlite3 connection context managers alone are not a general connection-closing guarantee. The lab also verifies disposal of its SQLAlchemy engines and keeps every file under an owned temporary workspace. Those checks establish this bounded exercise, not an operational restoration promise for another service or database dialect.

Practice solution and acceptance checks

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"))
                total = connection.scalar(text("SELECT coalesce(sum(score), 0) FROM roles"))
            return {"upgraded": upgraded, "changedScores": changed, "restoredRevision": revision,
                    "restoredScores": restored, "restoredHasLabel": has_label, "restoredTotal": total}
        finally:
            engine.dispose()
Reasoning
The solution queries the restored database for its total and returns ten for [3, 7], minus two for [0, -2], and zero for no rows. The rest of the recovery report remains unchanged. An omitted restore cannot satisfy the complete observed state merely by copying a total from the input. All database handles in the copy helper are explicitly closed; sqlite3 connection context managers alone are not a general connection-closing guarantee. The lab also verifies disposal of its SQLAlchemy engines and keeps every file under an owned temporary workspace. Those checks establish this bounded exercise, not an operational restoration promise for another service or database dialect.
Download the SQLAlchemy and Alembic computer lab ↗

Interview reasoning and next step

Present the complete evidence chain: the base revision and committed rows, an actual backup, the later migrated state, the restoration call, and independent reopened checks of schema, version and data.

Present the complete evidence chain: the base revision and committed rows, an actual backup, the later migrated state, the restoration call, and independent reopened checks of schema, version and data. Explain precisely which later writes the chosen image does not retain and why downgrade is a different operation.

Review the whole course through that chain: parameterized SQL, mapped constraints, transaction ownership, result shape, bounded queries, loading, set-based writes, savepoints, stale versions, async IO and reviewed migrations each protect a different contract. Finish by changing one controlled defect and naming which independent observation would expose it before copying this design into another environment.

Optional stretch — after the core task works: Design a recovery check for writes accepted after a backup and state what additional journal or reconciliation evidence it requires.

Computer practice: https://applydjinn.com/typorithm/sqlalchemy-lab-20261107.zip. Install the pinned requirements in an isolated Python 3.13 environment and run verify.py. Edit the working exercise baseline, then run --practice sa-capstone. The lab uses disposable SQLite databases and actual SQLAlchemy/Alembic calls; it does not connect to application accounts, production databases or paid services. The app includes prepared results and explained questions without an editor or Python runtime.

Verify both revision history and data after a real SQLite backup and restoration.