chore(api): transfer Alembic object ownership

docs/runbooks/strr-api.md

@@ -42,6 +42,7 @@ Flask backend for STRR. Source: `strr-api/`.
 - When `POD_NAMESPACE == "migration"`, the app factory registers **only** Flask-Migrate (no normal HTTP routes) (`strr-api/src/strr_api/__init__.py`).
 - **Driver note:** Migrations may use `postgresql+pg8000` while runtime uses `postgresql`/psycopg2. See `config.py` and `migrations/env.py`.
 - Standalone Alembic (no Flask context) uses `DATABASE_URL` with `NullPool` in `migrations/env.py`.
+- Optional ownership handoff: set `DATABASE_OWNER_ROLE` so online Alembic migrations run with that DB role and new objects are owned by it. The migration user must be able to `SET ROLE`, and that role needs schema create privileges.
 
 ## Scaling (prod)
 

strr-api/.env.sample

@@ -30,6 +30,9 @@ DATABASE_PASSWORD=postgres
 DATABASE_NAME=postgres
 DATABASE_HOST=localhost
 DATABASE_PORT=15432
+# Optional: Alembic runs migrations with this DB role so new objects are owned by it.
+# For Cloud SQL IAM auth, set this to the service account's Postgres role name.
+DATABASE_OWNER_ROLE=
 
 ## TEST DB
 DATABASE_TEST_USERNAME=postgres

strr-api/migrations/env.py

@@ -1,8 +1,9 @@
-import os
 import logging
+import os
 from logging.config import fileConfig
-from sqlalchemy import engine_from_config, pool, create_engine
+
 from alembic import context
+from sqlalchemy import create_engine, pool, text
 
 # 1. Passive Detection: Don't create an app, just look for one
 try:
@@ -40,14 +41,22 @@ def get_engine_url():
 def get_metadata():
     return target_metadata
 
+
 config.set_main_option('sqlalchemy.url', get_engine_url())
 
+
 def run_migrations_online():
     connectable = get_engine()
     with connectable.connect() as connection:
+        owner_role = os.getenv("DATABASE_OWNER_ROLE")
+        if owner_role:
+            safe_role = owner_role.replace('"', '""')  # Escape any quotes for SQL safety
+            connection.execute(text(f'SET ROLE "{safe_role}"'))
+            connection.commit()
+
         # Only pull migrate args if Flask is actually present
         extra_args = current_app.extensions['migrate'].configure_args if use_flask else {}
-        
+
         context.configure(
             connection=connection,
             target_metadata=get_metadata(),

strr-api/tests/integration/test_alembic_ownership.py

@@ -0,0 +1,66 @@
+"""Integration tests for standalone Alembic migration behavior."""
+
+from pathlib import Path
+
+from alembic import command
+from alembic.config import Config
+from sqlalchemy import create_engine, text
+from testcontainers.postgres import PostgresContainer
+
+
+def test_alembic_runs_with_configured_owner_role(monkeypatch):
+    """Alembic runs migrations with the configured DB owner role."""
+    api_root = Path(__file__).resolve().parents[2]
+    migrations_path = api_root / "migrations"
+    owner = "sa-api@bcrbk9-test.iam"
+
+    with PostgresContainer("postgres:16-alpine") as postgres:
+        db_url = postgres.get_connection_url()
+        monkeypatch.setenv("DATABASE_URL", db_url)
+        monkeypatch.setenv("DATABASE_OWNER_ROLE", owner)
+
+        engine = create_engine(db_url)
+        with engine.begin() as conn:
+            quoted_owner = conn.dialect.identifier_preparer.quote(owner)
+            conn.execute(text(f"CREATE ROLE {quoted_owner} LOGIN"))
+            conn.execute(text(f"GRANT USAGE, CREATE ON SCHEMA public TO {quoted_owner}"))
+
+        cfg = Config(str(migrations_path / "alembic.ini"))
+        cfg.set_main_option("script_location", str(migrations_path))
+        cfg.set_main_option("sqlalchemy.url", db_url)
+        command.upgrade(cfg, "head")
+
+        with engine.connect() as conn:
+            class_mismatches = conn.execute(
+                text(
+                    """
+                    SELECT c.relkind, n.nspname, c.relname, pg_get_userbyid(c.relowner) AS owner
+                    FROM pg_class c
+                    JOIN pg_namespace n ON n.oid = c.relnamespace
+                    WHERE n.nspname = 'public'
+                      AND c.relkind IN ('r', 'p', 'S', 'v', 'm', 'f')
+                      AND pg_get_userbyid(c.relowner) != :owner
+                    ORDER BY c.relkind, c.relname
+                    """
+                ),
+                {"owner": owner},
+            ).fetchall()
+            type_mismatches = conn.execute(
+                text(
+                    """
+                    SELECT n.nspname, t.typname, pg_get_userbyid(t.typowner) AS owner
+                    FROM pg_type t
+                    JOIN pg_namespace n ON n.oid = t.typnamespace
+                    WHERE n.nspname = 'public'
+                      AND t.typtype IN ('d', 'e')
+                      AND pg_get_userbyid(t.typowner) != :owner
+                    ORDER BY t.typname
+                    """
+                ),
+                {"owner": owner},
+            ).fetchall()
+            current_role = conn.execute(text("SELECT current_user")).scalar_one()
+
+        assert class_mismatches == []
+        assert type_mismatches == []
+        assert current_role != owner