Implement backfill for legacy WaterLevelsContinuous_Pressure_Daily table

.github/workflows/CD_staging.yml

@@ -37,6 +37,26 @@ jobs:
           with:
             credentials_json: ${{ secrets.CLOUD_DEPLOY_SERVICE_ACCOUNT_KEY }}
 
+        - name: Run Alembic migrations on staging database
+          env:
+            DB_DRIVER: "cloudsql"
+            CLOUD_SQL_INSTANCE_NAME: "${{ secrets.CLOUD_SQL_INSTANCE_NAME }}"
+            CLOUD_SQL_DATABASE: "${{ vars.CLOUD_SQL_DATABASE }}"
+            CLOUD_SQL_USER: "${{ secrets.CLOUD_SQL_USER }}"
+            CLOUD_SQL_PASSWORD: "${{ secrets.CLOUD_SQL_PASSWORD }}"
+          run: |
+            uv run alembic upgrade head
+
+        - name: Run backfill script on staging database
+          env:
+            DB_DRIVER: "cloudsql"
+            CLOUD_SQL_INSTANCE_NAME: "${{ secrets.CLOUD_SQL_INSTANCE_NAME }}"
+            CLOUD_SQL_DATABASE: "${{ vars.CLOUD_SQL_DATABASE }}"
+            CLOUD_SQL_USER: "${{ secrets.CLOUD_SQL_USER }}"
+            CLOUD_SQL_PASSWORD: "${{ secrets.CLOUD_SQL_PASSWORD }}"
+          run: |
+            uv run python transfers/backfill/staging.py
+
         # Uses Google Cloud Secret Manager to store secret credentials
         - name: Create app.yaml
           run: |

alembic/env.py

@@ -1,9 +1,9 @@
+import os
+from logging.config import fileConfig
+
 from alembic import context
 from dotenv import load_dotenv
-from logging.config import fileConfig
-from os import environ
-from sqlalchemy import engine_from_config
-from sqlalchemy import pool
+from sqlalchemy import engine_from_config, pool, create_engine
 
 
 # this is the Alembic Config object, which provides
@@ -33,15 +33,33 @@
 
 load_dotenv()
 
-# Fallback to environment variables for PostgreSQL connection
-user = environ.get("POSTGRES_USER", None)
-password = environ.get("POSTGRES_PASSWORD", None)
-db = environ.get("POSTGRES_DB", None)
-host = environ.get("POSTGRES_HOST", "localhost")
-port = environ.get("POSTGRES_PORT", 5432)
-SQLALCHEMY_DATABASE_URL = f"postgresql+psycopg2://{user}:{password}@{host}:{port}/{db}"
 
-config.set_main_option("sqlalchemy.url", SQLALCHEMY_DATABASE_URL)
+def build_database_url():
+    """
+    Build a SQLAlchemy URL based on driver/env vars.
+    For cloudsql we still return a pg8000 URL (hostless) so Alembic can render
+    offline migrations; the actual connection uses a Connector creator in
+    run_migrations_online.
+    """
+    db_driver = os.environ.get("DB_DRIVER", "").lower()
+    if db_driver == "cloudsql":
+        user = os.environ.get("CLOUD_SQL_USER", "")
+        password = os.environ.get("CLOUD_SQL_PASSWORD", "")
+        database = os.environ.get("CLOUD_SQL_DATABASE", "")
+        # Host is provided by connector, so leave blank.
+        return f"postgresql+pg8000://{user}:{password}@/{database}"
+
+    # Default/Postgres
+    user = os.environ.get("POSTGRES_USER", "")
+    password = os.environ.get("POSTGRES_PASSWORD", "")
+    db = os.environ.get("POSTGRES_DB", "")
+    host = os.environ.get("POSTGRES_HOST", "localhost")
+    port = os.environ.get("POSTGRES_PORT", 5432)
+    return f"postgresql+psycopg2://{user}:{password}@{host}:{port}/{db}"
+
+
+url = build_database_url()
+config.set_main_option("sqlalchemy.url", url)
 
 
 def include_object(object, name, type_, reflected, compare_to):
@@ -73,11 +91,41 @@ def run_migrations_online() -> None:
     and associate a connection with the context.
 
     """
-    connectable = engine_from_config(
-        config.get_section(config.config_ini_section, {}),
-        prefix="sqlalchemy.",
-        poolclass=pool.NullPool,
-    )
+    db_driver = os.environ.get("DB_DRIVER", "").lower()
+
+    if db_driver == "cloudsql":
+        # Use the Cloud SQL Python Connector for direct Cloud SQL access.
+        from google.cloud.sql.connector import Connector
+
+        instance_name = os.environ.get("CLOUD_SQL_INSTANCE_NAME")
+        user = os.environ.get("CLOUD_SQL_USER")
+        password = os.environ.get("CLOUD_SQL_PASSWORD")
+        database = os.environ.get("CLOUD_SQL_DATABASE")
+
+        connector = Connector()
+
+        def getconn():
+            return connector.connect(
+                instance_name,
+                "pg8000",
+                user=user,
+                password=password,
+                db=database,
+                ip_type="public",
+            )
+
+        connectable = create_engine(
+            "postgresql+pg8000://",
+            creator=getconn,
+            pool_pre_ping=True,
+            poolclass=pool.NullPool,
+        )
+    else:
+        connectable = engine_from_config(
+            config.get_section(config.config_ini_section, {}),
+            prefix="sqlalchemy.",
+            poolclass=pool.NullPool,
+        )
 
     with connectable.connect() as connection:
         context.configure(

alembic/versions/2101e0b029dc_make_location_description_nullable.py

@@ -0,0 +1,52 @@
+"""Make location description nullable
+
+Revision ID: 2101e0b029dc
+Revises: 66ac1af4ba69
+Create Date: 2026-01-02 23:19:38.901275
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import geoalchemy2
+import sqlalchemy as sa
+import sqlalchemy_utils
+
+
+def _column_exists(bind, table: str, column: str) -> bool:
+    inspector = sa.inspect(bind)
+    cols = [c["name"] for c in inspector.get_columns(table)]
+    return column in cols
+
+
+# revision identifiers, used by Alembic.
+revision: str = "2101e0b029dc"
+down_revision: Union[str, Sequence[str], None] = "66ac1af4ba69"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema.
+
+    Makes the location.description column nullable to accommodate
+    legacy data from MS Access that may not have descriptions.
+    """
+    bind = op.get_bind()
+    if _column_exists(bind, "location", "description"):
+        op.alter_column(
+            "location", "description", existing_type=sa.String(), nullable=True
+        )
+    else:
+        # If the column is absent (non-standard schema), skip the alteration.
+        pass
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    bind = op.get_bind()
+    if _column_exists(bind, "location", "description"):
+        op.alter_column(
+            "location", "description", existing_type=sa.String(), nullable=False
+        )