Complete worked example
python
from pathlib import Path
from tempfile import TemporaryDirectory
from sqlalchemy import create_engine, inspect, text
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 inspect_example(score):
with TemporaryDirectory() as folder:
config = configure_owned(folder)
engine = create_engine("sqlite+pysqlite://", connect_args={"autocommit": False})
try:
with engine.begin() as connection:
config.attributes["connection"] = connection
command.upgrade(config, "head")
head = MigrationContext.configure(connection).get_current_revision()
connection.execute(text("INSERT INTO roles (id, score) VALUES (1, :score)"), {"score": score})
command.downgrade(config, "owned_base")
current = MigrationContext.configure(connection).get_current_revision()
present = "roles" in inspect(connection).get_table_names()
columns = [c["name"] for c in inspect(connection).get_columns("roles")] if present else []
values = [list(row) for row in connection.execute(text("SELECT id, score FROM roles"))] if present else []
return {"head": head, "current": current, "columns": columns, "rows": values}
finally:
engine.dispose()- Reasoning
- command.upgrade(config, head) executes the pending revisions and records owned_label. The fixture then inserts a real row and downgrades to owned_base. The second revision’s downgrade removes only label, leaving the original id and score table definition and its row intact. The final report compares the head observed after upgrade with the actual current revision after downgrade, inspected columns and stored values. The env.py shown in the complete source consumes the supplied connection. Setting a connection attribute would not work by magic if the environment ignored that attribute and opened some other database.