← SQLAlchemy and Alembic: reliable database work Progress & review ↗

Typorithm · medium

SQLite batch changes that preserve rows and constraints

pythonsqlalchemyalembicdatabasesindependent-practice

Saved on this device

Full lesson · 4 learning steps Open any exercise for focused practice

Step 1 of 4 · intro card

SQLite batch changes that preserve rows and constraints

Concept and contract

Backfill existing values before tightening nullability and verify the recreated table’s constraints and indexes.

Some SQLite schema changes require a move-and-copy strategy rather than one portable ALTER statement. Alembic batch operations can reflect the old table, create a changed temporary table, copy rows, remove the original and rename the replacement. That is a real data-moving operation, not just editing a Python declaration.

The fixture starts with a nullable score and a named nonnegative-score CHECK. Existing NULL values need a policy before the score can become non-nullable. The lesson uses zero as that explicit owned-data policy and keeps the named CHECK through the table recreation.

Mechanism and worked example

The program inserts supplied scores, backfills NULL to zero and enters batch_alter_table with explicit recreation.

The program inserts supplied scores, backfills NULL to zero and enters batch_alter_table with explicit recreation. alter_column changes the score’s nullability while retaining its integer type. The final report reads actual stored scores, the inspected nullable flag, named CHECK constraints and indexes.

For [5, None, 2], the expected scores are [5, 0, 2], nullable is False and ck_roles_score still exists. These observations are independent: preserving values alone would not prove that the new column rule or named constraint survived, and a correct schema alone would not prove that rows were copied correctly.

Complete worked example

python

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, CheckConstraint, inspect, select, update
from alembic.migration import MigrationContext
from alembic.operations import Operations

def inspect_example(scores):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        metadata = MetaData()
        roles = Table("roles", metadata, Column("id", Integer, primary_key=True),
                      Column("score", Integer, nullable=True),
                      CheckConstraint("score >= 0", name="ck_roles_score"))
        metadata.create_all(engine)
        with engine.begin() as connection:
            if scores:
                connection.execute(roles.insert(), [{"id": i, "score": s} for i, s in enumerate(scores, 1)])
            connection.execute(update(roles).where(roles.c.score.is_(None)).values(score=0))
            operation = Operations(MigrationContext.configure(connection))
            with operation.batch_alter_table("roles", recreate="always") as batch:
                batch.alter_column("score", existing_type=Integer(), nullable=False)
            inspector = inspect(connection)
            score_column = next(c for c in inspector.get_columns("roles") if c["name"] == "score")
            return {"scores": list(connection.scalars(select(roles.c.score).order_by(roles.c.id))),
                    "nullable": score_column["nullable"],
                    "checks": sorted(c["name"] for c in inspector.get_check_constraints("roles")),
                    "indexes": sorted(i["name"] for i in inspector.get_indexes("roles"))}
    finally:
        engine.dispose()

Reasoning: The program inserts supplied scores, backfills NULL to zero and enters batch_alter_table with explicit recreation. alter_column changes the score’s nullability while retaining its integer type. The final report reads actual stored scores, the inspected nullable flag, named CHECK constraints and indexes. For [5, None, 2], the expected scores are [5, 0, 2], nullable is False and ck_roles_score still exists. These observations are independent: preserving values alone would not prove that the new column rule or named constraint survived, and a correct schema alone would not prove that rows were copied correctly.

Download the SQLAlchemy and Alembic computer lab ↗

Focused decision: context is in the complete example

python

with operation.batch_alter_table("roles", recreate="always") as batch:
    batch.alter_column("score", existing_type=Integer(), nullable=False)

Database decision: Rows are normalized before the stronger column rule is applied. Schema inspection verifies that the named CHECK and the added plain-column index survive the batch operation.

Download the SQLAlchemy and Alembic computer lab ↗

Counterexamples and common mistakes

The controlled mistake skips the data backfill.

The controlled mistake skips the data backfill. Copying a row that still has NULL into the new non-nullable table raises IntegrityError. The source schema looked reasonable, but the transition’s existing data could not satisfy its proposed destination.

Review unnamed constraints, referenced tables and index definitions when using recreation; they are not incidental decorations. This example uses a named CHECK and a plain-column index, not an expression-index behavior assumed from a newer unreleased fix. Locks, copy size and rollback characteristics also depend on the actual database, driver and transaction configuration.

Controlled bug: observe the rejected database contract

python

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, CheckConstraint, inspect, select, update
from alembic.migration import MigrationContext
from alembic.operations import Operations

def inspect_example(scores):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        metadata = MetaData()
        roles = Table("roles", metadata, Column("id", Integer, primary_key=True),
                      Column("score", Integer, nullable=True),
                      CheckConstraint("score >= 0", name="ck_roles_score"))
        metadata.create_all(engine)
        with engine.begin() as connection:
            if scores:
                connection.execute(roles.insert(), [{"id": i, "score": s} for i, s in enumerate(scores, 1)])
            # Missing the data backfill
            operation = Operations(MigrationContext.configure(connection))
            with operation.batch_alter_table("roles", recreate="always") as batch:
                batch.alter_column("score", existing_type=Integer(), nullable=False)
            inspector = inspect(connection)
            score_column = next(c for c in inspector.get_columns("roles") if c["name"] == "score")
            return {"scores": list(connection.scalars(select(roles.c.score).order_by(roles.c.id))),
                    "nullable": score_column["nullable"],
                    "checks": sorted(c["name"] for c in inspector.get_check_constraints("roles")),
                    "indexes": sorted(i["name"] for i in inspector.get_indexes("roles"))}
    finally:
        engine.dispose()

Reasoning: The controlled mistake skips the data backfill. Copying a row that still has NULL into the new non-nullable table raises IntegrityError. The source schema looked reasonable, but the transition’s existing data could not satisfy its proposed destination. Review unnamed constraints, referenced tables and index definitions when using recreation; they are not incidental decorations. This example uses a named CHECK and a plain-column index, not an expression-index behavior assumed from a newer unreleased fix. Locks, copy size and rollback characteristics also depend on the actual database, driver and transaction configuration.

Download the SQLAlchemy and Alembic computer lab ↗

Independent practice

First type the batch context and explicit nullable=False change, then predict why the backfill must precede it.

First type the batch context and explicit nullable=False change, then predict why the backfill must precede it. For independent practice add a named plain-column index ix_roles_score inside the same batch. Keep the existing CHECK, normalized values and non-nullable rule unchanged.

Add the index operation at the batch’s reviewed boundary rather than faking its name in the returned list. Use a mixed NULL fixture, an all-NULL fixture and an empty table. The empty case should still produce the requested schema even though there are no rows to transform.

Step 1: Create a named plain-column index on score inside the batch and verify the existing CHECK survived. Step 2: Preserve the existing CHECK, backfill and NOT NULL rule.

Practice starter: solve before reading the solution

python

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, CheckConstraint, inspect, select, update
from alembic.migration import MigrationContext
from alembic.operations import Operations

def inspect_example(scores):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        metadata = MetaData()
        roles = Table("roles", metadata, Column("id", Integer, primary_key=True),
                      Column("score", Integer, nullable=True),
                      CheckConstraint("score >= 0", name="ck_roles_score"))
        metadata.create_all(engine)
        with engine.begin() as connection:
            if scores:
                connection.execute(roles.insert(), [{"id": i, "score": s} for i, s in enumerate(scores, 1)])
            connection.execute(update(roles).where(roles.c.score.is_(None)).values(score=0))
            operation = Operations(MigrationContext.configure(connection))
            with operation.batch_alter_table("roles", recreate="always") as batch:
                batch.alter_column("score", existing_type=Integer(), nullable=False)
            inspector = inspect(connection)
            score_column = next(c for c in inspector.get_columns("roles") if c["name"] == "score")
            return {"scores": list(connection.scalars(select(roles.c.score).order_by(roles.c.id))),
                    "nullable": score_column["nullable"],
                    "checks": sorted(c["name"] for c in inspector.get_check_constraints("roles")),
                    "indexes": sorted(i["name"] for i in inspector.get_indexes("roles"))}
    finally:
        engine.dispose()

# Task: Create a named plain-column index on score inside the batch and verify the existing CHECK survived.

Reasoning: Create a named plain-column index on score inside the batch and verify the existing CHECK survived.

Download the SQLAlchemy and Alembic computer lab ↗

Hint 1: Add create_index inside the batch context after the column change.
Hint 2: Inspect the actual index list, including the empty-table case.

Solution, tests and trade-offs

The solution adds batch.create_index for score.

The solution adds batch.create_index for score. The inspector must now report ix_roles_score alongside the retained named CHECK, while row values and nullability remain identical to the worked contract. The tests inspect the actual database definitions instead of accepting source text that merely mentions an index.

The lab verifies this operation on its pinned SQLite stack. Creating an index is not evidence that a particular production query planner will choose it or that a large table can be rebuilt without unacceptable blocking. Those are separate workload and operational questions; the exercise establishes the requested schema object and preserved owned data.

Practice solution and acceptance checks

python

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, CheckConstraint, inspect, select, update
from alembic.migration import MigrationContext
from alembic.operations import Operations

def inspect_example(scores):
    engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
    try:
        metadata = MetaData()
        roles = Table("roles", metadata, Column("id", Integer, primary_key=True),
                      Column("score", Integer, nullable=True),
                      CheckConstraint("score >= 0", name="ck_roles_score"))
        metadata.create_all(engine)
        with engine.begin() as connection:
            if scores:
                connection.execute(roles.insert(), [{"id": i, "score": s} for i, s in enumerate(scores, 1)])
            connection.execute(update(roles).where(roles.c.score.is_(None)).values(score=0))
            operation = Operations(MigrationContext.configure(connection))
            with operation.batch_alter_table("roles", recreate="always") as batch:
                batch.alter_column("score", existing_type=Integer(), nullable=False)
                batch.create_index("ix_roles_score", ["score"])
            inspector = inspect(connection)
            score_column = next(c for c in inspector.get_columns("roles") if c["name"] == "score")
            return {"scores": list(connection.scalars(select(roles.c.score).order_by(roles.c.id))),
                    "nullable": score_column["nullable"],
                    "checks": sorted(c["name"] for c in inspector.get_check_constraints("roles")),
                    "indexes": sorted(i["name"] for i in inspector.get_indexes("roles"))}
    finally:
        engine.dispose()

Reasoning: The solution adds batch.create_index for score. The inspector must now report ix_roles_score alongside the retained named CHECK, while row values and nullability remain identical to the worked contract. The tests inspect the actual database definitions instead of accepting source text that merely mentions an index. The lab verifies this operation on its pinned SQLite stack. Creating an index is not evidence that a particular production query planner will choose it or that a large table can be rebuilt without unacceptable blocking. Those are separate workload and operational questions; the exercise establishes the requested schema object and preserved owned data.

Download the SQLAlchemy and Alembic computer lab ↗

Interview reasoning and next step

Describe the copy-and-replace phases and identify the point where old NULL values violate the destination.

Describe the copy-and-replace phases and identify the point where old NULL values violate the destination. Explain why an index name in a return dictionary is weaker evidence than database inspection, and why constraint names make retained semantics easier to verify during recreation.

Next, separate a schema expansion, a bounded data backfill and a later contract tightening. A small fixture can perform these in one owned transaction, but a deployed application may need separate releases and a resumable backfill. Understanding the ordering comes before deciding the operational rollout strategy.

Optional stretch — after the core task works: List the assumptions needed to recreate a larger referenced table on a chosen production dialect.

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-batch. 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.

Backfill existing values before tightening nullability and verify the recreated table’s constraints and indexes.