add alembic

container/Containerfile

@@ -26,8 +26,10 @@ RUN --mount=type=cache,target=/root/.cache/uv \
 
 FROM python:3.14-slim
 
+WORKDIR /app
 # Copy the environment, but not the source code
 COPY --from=builder /app/.venv /app/.venv
+COPY --from=builder /app/pyproject.toml /app/pyproject.toml
 
 # Copy the configuration files
 COPY config/lclstream_api.json  /etc/lclstream_api.json

pyproject.toml

@@ -25,6 +25,7 @@ dependencies = [
     "sqlalchemy>=2.0.51",
     "httpx>=0.28.1",
     "dbos>=2.0",
+    "alembic>=1.18.5",
 ]
 
 [dependency-groups]
@@ -37,6 +38,7 @@ dev = [
     "pytest-asyncio>=1.4.0",
     "ruff>=0.15.17",
     "types-pyyaml>=6.0.12.20260518",
+    "testcontainers[postgres]>=4.14.2",
 ]
 
 [build-system]
@@ -50,6 +52,21 @@ include-package-data = false
 where = ["src"]
 #include = ["lclstream_api*"]
 
+[tool.alembic]
+script_location = "lclstream_api.v2:alembic"
+
+[[tool.alembic.post_write_hooks]]
+name = "ruff_format"
+type = "exec"
+executable = "uv"
+options = "run ruff format REVISION_SCRIPT_FILENAME"
+
+[[tool.alembic.post_write_hooks]]
+name = "ruff_check"
+type = "exec"
+executable = "uv"
+options = "run ruff check --fix REVISION_SCRIPT_FILENAME"
+
 [tool.ruff]
 line-length = 88
 target-version = "py314"

scripts/gen_migration.py

@@ -0,0 +1,50 @@
+"""Generate an Alembic migration from SQLAlchemy ORM metadata."""
+
+import sys
+from pathlib import Path
+
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import create_engine
+from testcontainers.postgres import PostgresContainer
+
+import lclstream_api.v2.tables as _tables  # noqa: F401 — registers tables on Base.metadata
+
+POSTGRES_IMAGE = "postgres:18"
+POSTGRES_DRIVER = "psycopg"
+REPO_ROOT = Path(__file__).parent.parent
+VERSIONS_DIR = REPO_ROOT / "src/lclstream_api/v2/alembic/versions"
+
+
+def main() -> None:
+    message = (
+        " ".join(sys.argv[1:])
+        if len(sys.argv) > 1
+        else input("Migration name: ").strip()
+    )
+    if not message:
+        print("Aborted: migration name cannot be empty.")
+        sys.exit(1)
+
+    before = {f for f in VERSIONS_DIR.glob("*.py") if f.name != "__init__.py"}
+
+    print(f"Starting {POSTGRES_IMAGE}...")
+    with PostgresContainer(POSTGRES_IMAGE) as pg:
+        engine = create_engine(pg.get_connection_url(driver=POSTGRES_DRIVER))
+        cfg = Config(toml_file=str(REPO_ROOT / "pyproject.toml"))
+        cfg.set_main_option(
+            "sqlalchemy.url", engine.url.render_as_string(hide_password=False)
+        )
+        command.upgrade(cfg, "head")
+        command.revision(cfg, autogenerate=True, message=message)
+        engine.dispose()
+
+    new_files = {
+        f for f in VERSIONS_DIR.glob("*.py") if f.name != "__init__.py"
+    } - before
+    for f in sorted(new_files):
+        print(f"Generated: {f.name}")
+
+
+if __name__ == "__main__":
+    main()

src/lclstream_api/v2/alembic/env.py

@@ -0,0 +1,34 @@
+import logging
+
+from alembic import context
+from sqlalchemy import create_engine, pool
+
+import lclstream_api.v2.tables as _tables  # noqa: F401 - registers tables
+from lclstream_api.v2.config import database
+from lclstream_api.v2.tables import Base
+
+config = context.config
+logging.basicConfig(level=logging.INFO)
+target_metadata = Base.metadata
+
+
+def get_url() -> str:
+    # set programmatically by scripts/gen_migration.py for testcontainers
+    url = config.get_alembic_option("sqlalchemy.url")
+    if url:
+        return url
+    # for running migrations against real db
+    return str(database.url)
+
+
+def run_migrations() -> None:
+    connectable = create_engine(get_url(), poolclass=pool.NullPool)
+
+    with connectable.connect() as connection:
+        context.configure(connection=connection, target_metadata=target_metadata)
+
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+run_migrations()

src/lclstream_api/v2/alembic/script.py.mako

@@ -0,0 +1,24 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+
+"""
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+# revision identifiers, used by Alembic.
+revision: str = ${repr(up_revision)}
+down_revision: str | list[str] | None = ${repr(down_revision)}
+branch_labels: str | list[str] | None = ${repr(branch_labels)}
+depends_on: str | list[str] | None = ${repr(depends_on)}
+
+
+def upgrade():
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade():
+    ${downgrades if downgrades else "pass"}

src/lclstream_api/v2/alembic/versions/6ead17ed69da_init.py

@@ -0,0 +1,70 @@
+"""init
+
+Revision ID: 6ead17ed69da
+Revises:
+Create Date: 2026-07-08 10:49:38.890905
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "6ead17ed69da"
+down_revision: str | list[str] | None = None
+branch_labels: str | list[str] | None = None
+depends_on: str | list[str] | None = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "transfers",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("user", sa.String(), nullable=False),
+        sa.Column("state", sa.String(), nullable=False),
+        sa.Column("parameters", sa.JSON(), nullable=False),
+        sa.Column("cache_id", sa.Uuid(), nullable=True),
+        sa.Column("cache_hostname", sa.String(), nullable=True),
+        sa.Column("pull_port", sa.Integer(), nullable=True),
+        sa.Column("push_port", sa.Integer(), nullable=True),
+        sa.Column("producer_job_id", sa.String(), nullable=True),
+        sa.Column("last_polled_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_table(
+        "transitions",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("transfer_id", sa.Uuid(), nullable=False),
+        sa.Column("state", sa.String(), nullable=False),
+        sa.Column("info", sa.String(), nullable=True),
+        sa.Column("source", sa.String(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.ForeignKeyConstraint(["transfer_id"], ["transfers.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("transitions")
+    op.drop_table("transfers")
+    # ### end Alembic commands ###

src/lclstream_api/v2/db.py

@@ -24,7 +24,6 @@
 )
 
 from .config import database
-from .tables import Base
 
 engine = create_async_engine(str(database.url))
 async_session = async_sessionmaker(engine, expire_on_commit=False)
@@ -35,9 +34,6 @@
 async def init_datasource() -> AsyncSQLAlchemyDatasource:
     global _datasource
     ds = await AsyncSQLAlchemyDatasource.create(str(database.url), engine=engine)
-    # create tables
-    async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.create_all)
     _datasource = ds
     return ds
 

tests/v2/test_migrations.py

@@ -0,0 +1,49 @@
+from collections.abc import Generator
+from pathlib import Path
+
+import pytest
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import Engine, create_engine
+from testcontainers.postgres import PostgresContainer
+
+import lclstream_api.v2.tables as _tables  # noqa: F401 — registers all tables in Base.metadata
+
+POSTGRES_IMAGE = "postgres:18"
+POSTGRES_DRIVER = "psycopg"
+REPO_ROOT = Path(__file__).parent.parent.parent
+
+
+def _alembic_config(engine: Engine) -> Config:
+    cfg = Config(toml_file=str(REPO_ROOT / "pyproject.toml"))
+    cfg.set_main_option(
+        "sqlalchemy.url", engine.url.render_as_string(hide_password=False)
+    )
+    return cfg
+
+
+@pytest.fixture(scope="module")
+def postgres_engine() -> Generator[Engine]:
+    """Spin up a PostgreSQL container, apply all migrations, and yield a connected engine."""
+    with PostgresContainer(POSTGRES_IMAGE) as pg:
+        engine = create_engine(pg.get_connection_url(driver=POSTGRES_DRIVER))
+        command.upgrade(_alembic_config(engine), "head")
+        yield engine
+        engine.dispose()
+
+
+class TestMigrations:
+    def test_upgrade_succeeds(self, postgres_engine: Engine) -> None:
+        """alembic upgrade head must complete without error."""
+
+    def test_no_model_drift(self, postgres_engine: Engine) -> None:
+        """alembic check: no model changes are missing from the migration history."""
+        cfg = _alembic_config(postgres_engine)
+        # command.check raises MigrationSchemaMismatch if drift is detected.
+        command.check(cfg)
+
+    def test_downgrade_and_upgrade(self, postgres_engine: Engine) -> None:
+        """All migrations must be reversible back to base and re-applicable to head."""
+        cfg = _alembic_config(postgres_engine)
+        command.downgrade(cfg, "base")
+        command.upgrade(cfg, "head")