feat(tenancy): route each signed-in request to its tenant's schema (M2b, #93)

alembic/versions/0004_control_schema.py

@@ -8,7 +8,7 @@
 Create Date: 2026-10-26
 """
 
-from alembic import op
+from alembic import context, op
 
 from api.config import settings
 
@@ -22,7 +22,14 @@ def _schema() -> str:
     return op.get_bind().dialect.identifier_preparer.quote(settings.CONTROL_SCHEMA)
 
 
+def _tenant_run() -> bool:
+    # Shared objects are created once, not in every tenant schema (api/tenancy.py).
+    return context.config.attributes.get("scope") == "tenant"
+
+
 def upgrade() -> None:
+    if _tenant_run():
+        return
     s = _schema()
     op.execute(f"CREATE SCHEMA IF NOT EXISTS {s}")
     op.execute(f"""
@@ -59,4 +66,6 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
+    if _tenant_run():
+        return
     op.execute(f"DROP SCHEMA IF EXISTS {_schema()} CASCADE")

api/auth.py

@@ -12,12 +12,15 @@
 """
 
 import secrets
+from collections.abc import AsyncIterator
 
 from fastapi import Depends, Header, HTTPException
+from starlette.concurrency import run_in_threadpool
 
 from api.config import settings
 from api.db import Conn
 from api.services.accounts import session_for
+from api.tenancy import schema_for, scoped
 
 
 def require_api_key(x_api_key: str = Header(default="")) -> None:
@@ -39,11 +42,34 @@ def bearer_token(authorization: str = Header(default="")) -> str:
 def require_session(token: str = Depends(bearer_token)) -> dict:
     """The caller's live session (user, tenant, role), or 401 (M2a, #92).
 
-    Not yet applied to the data routers; tenant routing (#93) and the role
-    matrix (#94) switch them over.
+    Data routers use ``tenant_scope``, which also routes to the tenant's schema.
     """
+    return _live_session(token)
+
+
+def _live_session(token: str) -> dict:
     with Conn() as conn, conn.cursor() as cur:
         session = session_for(cur, token)
     if session is None:
         raise HTTPException(status_code=401, detail="Session expired or invalid; sign in again")
     return session
+
+
+async def tenant_scope(authorization: str = Header(default=""),
+                       x_api_key: str = Header(default="")) -> AsyncIterator[dict | None]:
+    """Route the request to one tenant's schema (M2b, #93).
+
+    With ``Authorization: Bearer <token>`` the request runs in the signed-in
+    tenant's schema and yields its session. Without one, the shared API key
+    applies and the request keeps the deployment's default schemas.
+
+    Async on purpose: the schema is a context variable, and only an async
+    dependency sets it in the context the endpoint's worker thread copies.
+    """
+    if not authorization:
+        require_api_key(x_api_key)
+        yield None
+        return
+    session = await run_in_threadpool(_live_session, bearer_token(authorization))
+    with scoped(schema_for(session["tenant"])):
+        yield session

api/db.py

@@ -159,8 +159,7 @@ def bootstrap() -> None:
     from alembic import command
     from alembic.config import Config
 
-    repo_root = Path(__file__).resolve().parent.parent
-    alembic_ini = repo_root / "alembic.ini"
+    alembic_ini = alembic_ini_path()
 
     if not alembic_ini.exists():
         logger.warning(
@@ -177,9 +176,28 @@ def bootstrap() -> None:
                 settings.PG_USER, settings.PG_HOST, settings.PG_PORT,
                 settings.PG_DATABASE)
     command.upgrade(cfg, "head")
+    _migrate_existing_tenants()
     logger.info("Alembic migrations complete")
 
 
+def alembic_ini_path() -> Path:
+    return Path(__file__).resolve().parent.parent / "alembic.ini"
+
+
+def _migrate_existing_tenants() -> None:
+    """Bring every tenant schema to head too (decision 6: migrations run across N schemas)."""
+    from api.tenancy import migrate_tenant
+
+    with Conn() as conn, conn.cursor() as cur:
+        cur.execute(sql.SQL("SELECT slug FROM {}.tenants ORDER BY id").format(
+            sql.Identifier(settings.CONTROL_SCHEMA)))
+        slugs = [r[0] for r in cur.fetchall()]
+    for slug in slugs:
+        migrate_tenant(slug)
+    if slugs:
+        logger.info("Migrated %d tenant schema(s)", len(slugs))
+
+
 def _bootstrap_fallback_schema() -> None:
     """Idempotent CREATE TABLE IF NOT EXISTS bootstrap used when Alembic isn't
     wired up (see bootstrap() above). Mirrors the columns every INSERT/SELECT

api/routers/audit_routes.py

@@ -12,14 +12,14 @@
 from fastapi import APIRouter, Depends
 from psycopg2 import sql
 
-from api.auth import require_api_key
-from api.config import settings
+from api.auth import tenant_scope
+from api.tenancy import uploads_schema
 from api.db import Conn
 
 router = APIRouter(
     prefix="/api/audit",
     tags=["audit"],
-    dependencies=[Depends(require_api_key)],
+    dependencies=[Depends(tenant_scope)],
 )
 
 
@@ -31,15 +31,15 @@ def audit_log(limit: int = 50, offset: int = 0) -> dict:
         with conn.cursor() as cur:
             cur.execute(
                 sql.SQL("SELECT COUNT(*) FROM {}.audit_log").format(
-                    sql.Identifier(settings.UPLOADS_SCHEMA)
+                    sql.Identifier(uploads_schema())
                 )
             )
             total = cur.fetchone()[0]
             cur.execute(
                 sql.SQL(
                     "SELECT id, action, file_id, file_name, table_name, mode, performed_at "
                     "FROM {}.audit_log ORDER BY performed_at DESC LIMIT %s OFFSET %s"
-                ).format(sql.Identifier(settings.UPLOADS_SCHEMA)),
+                ).format(sql.Identifier(uploads_schema())),
                 (limit, offset),
             )
             entries = [

api/routers/csv_routes.py

@@ -8,8 +8,9 @@
 from psycopg2 import sql
 from pydantic import BaseModel, Field
 
-from api.auth import require_api_key
+from api.auth import tenant_scope
 from api.config import settings
+from api.tenancy import current_schema, uploads_schema
 from api.db import Conn
 from api.services.audit import coerce_file_id, record_audit, write_audit
 from api.services.csv_parse import build_preview
@@ -23,7 +24,7 @@
 router = APIRouter(
     prefix="/api/csv",
     tags=["csv"],
-    dependencies=[Depends(require_api_key)],
+    dependencies=[Depends(tenant_scope)],
 )
 
 
@@ -71,6 +72,9 @@ def upload(req: UploadRequest) -> dict:
     if len(req.content.encode("utf-8")) > settings.MAX_UPLOAD_BYTES:
         raise HTTPException(413, "File too large")
     if req.mode == "te":
+        if current_schema() is not None:
+            raise HTTPException(409, "T&E mode isn't available for tenants yet; it arrives with "
+                                     "per-tenant T&E tables (#99). Use dynamic mode.")
         if not req.targetTable:
             raise HTTPException(422, "targetTable is required when mode is 'te'")
         result = upload_te(req.fileName, req.content, req.targetTable)
@@ -103,7 +107,7 @@ def list_files() -> list[dict]:
                     "SELECT id, file_name, table_name, mode, row_count, column_names, created_at, "
                     "total_rows, duplicate_rows_skipped, failed_rows "
                     "FROM {}.csv_files ORDER BY created_at DESC, id DESC"
-                ).format(sql.Identifier(settings.UPLOADS_SCHEMA))
+                ).format(sql.Identifier(uploads_schema()))
             )
             return [
                 {
@@ -132,15 +136,15 @@ def table_rows(table_name: str, limit: int = 50) -> dict:
             # Only serve tables that are registered uploads
             cur.execute(
                 sql.SQL("SELECT 1 FROM {}.csv_files WHERE table_name = %s").format(
-                    sql.Identifier(settings.UPLOADS_SCHEMA)
+                    sql.Identifier(uploads_schema())
                 ),
                 (table_name,),
             )
             if cur.fetchone() is None:
                 raise HTTPException(404, "Table not found")
             cur.execute(
                 sql.SQL("SELECT * FROM {}.{} ORDER BY _id LIMIT %s").format(
-                    sql.Identifier(settings.UPLOADS_SCHEMA), sql.Identifier(table_name)
+                    sql.Identifier(uploads_schema()), sql.Identifier(table_name)
                 ),
                 (limit,),
             )
@@ -167,7 +171,7 @@ def delete_file(file_id: int) -> dict:
             cur.execute(
                 sql.SQL(
                     "SELECT table_name, mode, file_name FROM {}.csv_files WHERE id = %s"
-                ).format(sql.Identifier(settings.UPLOADS_SCHEMA)),
+                ).format(sql.Identifier(uploads_schema())),
                 (file_id,),
             )
             row = cur.fetchone()
@@ -181,18 +185,18 @@ def delete_file(file_id: int) -> dict:
                     SELECT 1 FROM information_schema.tables
                     WHERE table_schema = %s AND table_name = %s
                     """,
-                    (settings.UPLOADS_SCHEMA, table_name),
+                    (uploads_schema(), table_name),
                 )
                 if cur.fetchone() is not None:
                     cur.execute(
                         sql.SQL("DROP TABLE IF EXISTS {}.{}").format(
-                            sql.Identifier(settings.UPLOADS_SCHEMA),
+                            sql.Identifier(uploads_schema()),
                             sql.Identifier(table_name),
                         )
                     )
             cur.execute(
                 sql.SQL("DELETE FROM {}.csv_files WHERE id = %s").format(
-                    sql.Identifier(settings.UPLOADS_SCHEMA)
+                    sql.Identifier(uploads_schema())
                 ),
                 (file_id,),
             )

api/routers/te_routes.py

@@ -2,15 +2,16 @@
 
 from fastapi import APIRouter, Depends
 
-from api.auth import require_api_key
-from api.config import TE_TABLES, settings
+from api.auth import tenant_scope
+from api.config import TE_TABLES
+from api.tenancy import te_schema
 from api.db import Conn
 
 # BUG-035: auth is applied per-router (see csv_routes.py for the rationale).
 router = APIRouter(
     prefix="/api/te",
     tags=["te"],
-    dependencies=[Depends(require_api_key)],
+    dependencies=[Depends(tenant_scope)],
 )
 
 
@@ -30,15 +31,15 @@ def te_tables() -> list[dict]:
                 SELECT table_name FROM information_schema.tables
                 WHERE table_schema = %s AND table_name = ANY(%s)
                 """,
-                (settings.TE_SCHEMA, tables),
+                (te_schema(), tables),
             )
             existing = {r[0] for r in cur.fetchall()}
             cur.execute(
                 """
                 SELECT relname, n_live_tup FROM pg_stat_user_tables
                 WHERE schemaname = %s AND relname = ANY(%s)
                 """,
-                (settings.TE_SCHEMA, tables),
+                (te_schema(), tables),
             )
             row_counts = {r[0]: max(r[1], 0) for r in cur.fetchall()}
     return [

api/services/audit.py

@@ -28,7 +28,7 @@
 
 from psycopg2 import sql
 
-from api.config import settings
+from api.tenancy import uploads_schema
 from api.db import Conn
 
 logger = logging.getLogger(__name__)
@@ -52,7 +52,7 @@ def write_audit(cur, action: str, file_id: int | None, file_name: str | None,
             "INSERT INTO {}.audit_log "
             "(action, file_id, file_name, table_name, mode) "
             "VALUES (%s, %s, %s, %s, %s)"
-        ).format(sql.Identifier(settings.UPLOADS_SCHEMA)),
+        ).format(sql.Identifier(uploads_schema())),
         (action, file_id, file_name, table_name, mode),
     )
 

api/services/dynamic_loader.py

@@ -18,6 +18,7 @@
 from psycopg2.extras import execute_values
 
 from api.config import settings
+from api.tenancy import uploads_schema
 from api.db import Conn
 from api.services.csv_parse import (
     ALLOWED_TYPES,
@@ -103,7 +104,7 @@ def upload_dynamic(
             "logs": logs,
         }
 
-    schema = settings.UPLOADS_SCHEMA
+    schema = uploads_schema()
 
     try:
         return _do_upload(file_name, types, overwrite, logs, schema, rows, file_hash)