feat(db): Alembic migrations and registry UNIQUE constraints (#84, #81)

alembic.ini

@@ -0,0 +1,9 @@
+# Alembic configuration for the API's uploads schema (#84).
+# The database URL and target schema come from api/config.py (libpq env vars),
+# so there is no sqlalchemy.url here. Run from the repo root:
+#   alembic upgrade head
+# api/db.py bootstrap() runs the same upgrade on every API start.
+
+[alembic]
+script_location = %(here)s/alembic
+path_separator = os

alembic/env.py

@@ -0,0 +1,40 @@
+"""Alembic environment for the uploads schema.
+
+Connection settings come from api/config.py, the same values the API pool uses.
+The version table lives inside the uploads schema, so each schema carries its
+own migration state (this matters once there is one schema per tenant).
+"""
+
+import sys
+from pathlib import Path
+
+from alembic import context
+from sqlalchemy import create_engine, text
+from sqlalchemy.engine import URL
+from sqlalchemy.pool import NullPool
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from api.config import settings  # noqa: E402
+
+if context.is_offline_mode():
+    raise RuntimeError("Offline (--sql) migrations are not supported; run against a database.")
+
+url = URL.create(
+    "postgresql+psycopg2",
+    username=settings.PG_USER,
+    password=settings.PG_PASSWORD or None,
+    host=settings.PG_HOST,
+    port=settings.PG_PORT,
+    database=settings.PG_DATABASE,
+)
+schema = settings.UPLOADS_SCHEMA
+
+engine = create_engine(url, poolclass=NullPool)
+with engine.connect() as connection:
+    quoted = connection.dialect.identifier_preparer.quote(schema)
+    connection.execute(text(f"CREATE SCHEMA IF NOT EXISTS {quoted}"))
+    connection.commit()
+    context.configure(connection=connection, version_table_schema=schema)
+    with context.begin_transaction():
+        context.run_migrations()

alembic/script.py.mako

@@ -0,0 +1,21 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+"""
+
+from alembic import op
+
+revision = ${repr(up_revision)}
+down_revision = ${repr(down_revision)}
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}

alembic/versions/0001_baseline.py

@@ -0,0 +1,59 @@
+"""Baseline: csv_files registry and audit_log, as created by the fallback bootstrap.
+
+Uses IF NOT EXISTS so databases already created by api/db.py's fallback
+bootstrap adopt this revision without changes.
+
+Revision ID: 0001
+Revises:
+Create Date: 2026-10-26
+"""
+
+from alembic import op
+
+from api.config import settings
+
+revision = "0001"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def _schema() -> str:
+    return op.get_bind().dialect.identifier_preparer.quote(settings.UPLOADS_SCHEMA)
+
+
+def upgrade() -> None:
+    s = _schema()
+    op.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS {s}.csv_files (
+            id BIGSERIAL PRIMARY KEY,
+            file_name TEXT NOT NULL,
+            file_hash TEXT NOT NULL,
+            table_name TEXT NOT NULL,
+            mode TEXT NOT NULL,
+            row_count INTEGER NOT NULL DEFAULT 0,
+            column_names TEXT[] NOT NULL DEFAULT '{{}}',
+            created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+        )
+        """
+    )
+    op.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS {s}.audit_log (
+            id BIGSERIAL PRIMARY KEY,
+            action TEXT NOT NULL,
+            file_id BIGINT,
+            file_name TEXT,
+            table_name TEXT,
+            mode TEXT,
+            performed_at TIMESTAMPTZ NOT NULL DEFAULT now()
+        )
+        """
+    )
+
+
+def downgrade() -> None:
+    s = _schema()
+    op.execute(f"DROP TABLE IF EXISTS {s}.audit_log")
+    op.execute(f"DROP TABLE IF EXISTS {s}.csv_files")

alembic/versions/0002_registry_unique.py

@@ -0,0 +1,53 @@
+"""Registry uniqueness: one row per file hash and per file name (#81).
+
+dynamic_loader.py relies on these constraints to settle concurrent identical
+uploads, but they were never created. If a database already holds duplicates
+(possible because of that race), the upgrade stops and lists them instead of
+deleting anything.
+
+Note: UNIQUE (file_name) matches today's behaviour (a repeated filename is
+refused or replaced). ART decision 2 (versioned datasets) revisits it in M5.
+
+Revision ID: 0002
+Revises: 0001
+Create Date: 2026-10-26
+"""
+
+from alembic import op
+from sqlalchemy import text
+
+from api.config import settings
+
+revision = "0002"
+down_revision = "0001"
+branch_labels = None
+depends_on = None
+
+
+def _schema() -> str:
+    return op.get_bind().dialect.identifier_preparer.quote(settings.UPLOADS_SCHEMA)
+
+
+def upgrade() -> None:
+    s = _schema()
+    bind = op.get_bind()
+    for column in ("file_hash", "file_name"):
+        dupes = bind.execute(
+            # column comes from the fixed tuple above; s is quoted by the dialect.
+            text(f"SELECT {column}, count(*) FROM {s}.csv_files GROUP BY {column} HAVING count(*) > 1")  # nosec B608
+        ).fetchall()
+        if dupes:
+            listed = ", ".join(f"{value!r} x{count}" for value, count in dupes[:10])
+            raise RuntimeError(
+                f"Cannot add UNIQUE ({column}) to {settings.UPLOADS_SCHEMA}.csv_files: "
+                f"{len(dupes)} duplicated value(s): {listed}. Remove the duplicate registry "
+                f"rows (keep the one whose table_name still exists), then restart the API."
+            )
+    op.execute(f"ALTER TABLE {s}.csv_files ADD CONSTRAINT csv_files_file_hash_key UNIQUE (file_hash)")
+    op.execute(f"ALTER TABLE {s}.csv_files ADD CONSTRAINT csv_files_file_name_key UNIQUE (file_name)")
+
+
+def downgrade() -> None:
+    s = _schema()
+    op.execute(f"ALTER TABLE {s}.csv_files DROP CONSTRAINT IF EXISTS csv_files_file_name_key")
+    op.execute(f"ALTER TABLE {s}.csv_files DROP CONSTRAINT IF EXISTS csv_files_file_hash_key")

api/db.py

@@ -198,8 +198,8 @@ def _bootstrap_fallback_schema() -> None:
                     """
                     CREATE TABLE IF NOT EXISTS {}.csv_files (
                         id BIGSERIAL PRIMARY KEY,
-                        file_name TEXT NOT NULL,
-                        file_hash TEXT NOT NULL,
+                        file_name TEXT NOT NULL UNIQUE,
+                        file_hash TEXT NOT NULL UNIQUE,
                         table_name TEXT NOT NULL,
                         mode TEXT NOT NULL,
                         row_count INTEGER NOT NULL DEFAULT 0,

api/services/dynamic_loader.py

@@ -105,6 +105,19 @@ def upload_dynamic(
 
     try:
         return _do_upload(file_name, types, overwrite, logs, schema, rows, file_hash)
+    except psycopg2.errors.UniqueViolation:
+        # #81: an identical concurrent upload won a race that _do_upload could
+        # not settle itself (e.g. both creating csv_<hash> at once). Nothing
+        # from this request was committed.
+        _log(logs, "duplicate_check", "Lost race to a concurrent identical upload", "warn")
+        return {
+            "status": "duplicate_file",
+            "reason": "content",
+            "existingFileName": file_name,
+            "tableName": f"csv_{file_hash[:16]}",
+            "existingRowCount": 0,
+            "logs": logs,
+        }
     except psycopg2.Error as exc:
         # Mirror te_loader's structured-error contract: an unexpected DB error
         # (e.g. a column type edge case not caught by cast_value) should not

api/services/te_loader.py

@@ -22,6 +22,7 @@
 import hashlib
 import time
 
+import psycopg2.errors
 from psycopg2 import sql
 from psycopg2.extras import execute_values
 
@@ -290,13 +291,24 @@ def _err(row_number: int, column: str, value: str, reason: str) -> None:
             # *dynamic*-mode upload of the same file also has its csv_<hash>
             # table dropped instead of orphaned.
             _cleanup_prior_registry_entries(cur, file_name, file_hash)
-            cur.execute(
-                sql.SQL(
-                    "INSERT INTO {}.csv_files (file_name, file_hash, table_name, mode, row_count, column_names) "
-                    "VALUES (%s, %s, %s, 'te', %s, %s) RETURNING id"
-                ).format(sql.Identifier(settings.UPLOADS_SCHEMA)),
-                (file_name, file_hash, f"{settings.TE_SCHEMA}.{target_table}", inserted, columns),
-            )
+            try:
+                cur.execute(
+                    sql.SQL(
+                        "INSERT INTO {}.csv_files (file_name, file_hash, table_name, mode, row_count, column_names) "
+                        "VALUES (%s, %s, %s, 'te', %s, %s) RETURNING id"
+                    ).format(sql.Identifier(settings.UPLOADS_SCHEMA)),
+                    (file_name, file_hash, f"{settings.TE_SCHEMA}.{target_table}", inserted, columns),
+                )
+            except psycopg2.errors.UniqueViolation:
+                # #81: a concurrent load of the same file registered first.
+                # Roll back this load's rows too, so nothing lands twice.
+                conn.rollback()
+                _log(logs, "duplicate_check", "Lost race to a concurrent load of the same file", "warn")
+                return {
+                    "status": "error",
+                    "message": "The same file was loaded concurrently and that load finished first. Nothing was imported by this request.",
+                    "logs": logs,
+                }
             file_id = cur.fetchone()[0]
         conn.commit()
 

scripts/lint.sh

@@ -12,6 +12,10 @@ PY_SRC=(
     api/main.py
     api/ratelimit.py
     api/services/audit.py
+    api/db.py
+    alembic/env.py
+    alembic/versions/0001_baseline.py
+    alembic/versions/0002_registry_unique.py
 )
 
 PY_TESTS=(
@@ -23,6 +27,7 @@ PY_TESTS=(
     tests/test_issue_05_import_summary.py
     tests/test_api_hardening.py
     tests/test_evals_tier_e_x.py
+    tests/test_migrations.py
 )
 
 # Collect scripts/*.py dynamically