fix(ingestion): supply GCP credentials in a runtime that has none

automated_ingestion/defs/resources.py

@@ -39,6 +39,16 @@ class OcotilloDatabase(ConfigurableResource):
     @contextmanager
     def session(self) -> Iterator[object]:
         """Yield a SQLAlchemy session, rolled back and closed on the way out."""
+        # Credentials first: db.engine builds its Cloud SQL connector at import
+        # time, and the connector resolves Application Default Credentials right
+        # then. Serverless has none until they are written to disk, so doing this
+        # afterwards would be too late.
+        from automated_ingestion.shared.credentials import (
+            ensure_application_default_credentials,
+        )
+
+        ensure_application_default_credentials()
+
         # Imported lazily: importing db.engine builds an engine from the
         # environment at import time, which should happen when a run asks for a
         # session, not when Dagster loads the code location to list assets.

automated_ingestion/scripts/set_code_location_env.sh

@@ -21,6 +21,7 @@
 #
 # Usage:
 #     ./automated_ingestion/scripts/set_code_location_env.sh storage
+#     ./automated_ingestion/scripts/set_code_location_env.sh credentials
 #     ./automated_ingestion/scripts/set_code_location_env.sh vendor
 #     ./automated_ingestion/scripts/set_code_location_env.sh database
 #
@@ -53,6 +54,16 @@ storage)
   set_var INGESTION_GCS_BUCKET ocotillo-ingestion-production --scope full
   set_var INGESTION_GCS_BUCKET ocotillo-ingestion-staging --scope branch
   ;;
+credentials)
+  : "${INGESTION_GCP_CREDENTIALS_JSON:?export the service account key JSON, not a path}"
+  # Serverless runs outside GCP, so there is no metadata server and nothing
+  # supplies Application Default Credentials. Both the Cloud SQL connector and
+  # gcsfs need them. Mint the key with:
+  #   gcloud iam service-accounts keys create /dev/stdout \
+  #     --iam-account ocotillo-ingestion@waterdatainitiative-271000.iam.gserviceaccount.com
+  echo "GCP credentials (key JSON read from this shell, not echoed):"
+  set_var INGESTION_GCP_CREDENTIALS_JSON --from-local-env
+  ;;
 vendor)
   : "${DIVERHUB_USERNAME:?export it first, see the header}"
   : "${DIVERHUB_PASSWORD:?export it first, see the header}"
@@ -89,7 +100,7 @@ database)
   fi
   ;;
 *)
-  echo "usage: $0 {storage|vendor|database}" >&2
+  echo "usage: $0 {storage|credentials|vendor|database}" >&2
   exit 64
   ;;
 esac

automated_ingestion/shared/credentials.py

@@ -0,0 +1,83 @@
+# ===============================================================================
+# Copyright 2026 ross
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+"""
+Application Default Credentials for a runtime that has none.
+
+Dagster+ Serverless runs outside GCP, so there is no metadata server to supply
+credentials. Anything reaching Google -- the Cloud SQL connector for the loader,
+gcsfs for the raw zone -- calls ``google.auth.default()`` and fails with
+``DefaultCredentialsError`` unless something has put credentials on disk first.
+
+The service account key therefore travels as a Dagster+ secret and is written to
+a file here, because ``GOOGLE_APPLICATION_CREDENTIALS`` names a path rather than
+holding a value. The file lands in the process's temporary directory, which the
+container discards when the run ends.
+"""
+
+import json
+import os
+import tempfile
+
+CREDENTIALS_ENV_VAR = "INGESTION_GCP_CREDENTIALS_JSON"
+"""Service account key JSON, as a Dagster+ secret. Never committed."""
+
+_ADC_ENV_VAR = "GOOGLE_APPLICATION_CREDENTIALS"
+
+_written_path: str | None = None
+
+
+def ensure_application_default_credentials() -> str | None:
+    """Materialize ADC from the environment, returning the path if written.
+
+    Idempotent, and does nothing when credentials already exist -- locally that
+    means a developer's gcloud login is used as-is rather than being shadowed.
+    """
+    global _written_path
+
+    existing = os.environ.get(_ADC_ENV_VAR, "").strip()
+    if existing:
+        return existing
+    if _written_path is not None:
+        return _written_path
+
+    raw = os.environ.get(CREDENTIALS_ENV_VAR, "").strip()
+    if not raw:
+        # No key configured. Leave google.auth to its own discovery, which
+        # succeeds on a developer machine and fails loudly in Serverless -- the
+        # right outcome in both cases.
+        return None
+
+    try:
+        parsed = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise RuntimeError(
+            f"{CREDENTIALS_ENV_VAR} is set but is not valid JSON. It must hold the "
+            "service account key itself, not a path to one."
+        ) from exc
+
+    handle = tempfile.NamedTemporaryFile(
+        mode="w", suffix=".json", prefix="ingestion-adc-", delete=False
+    )
+    with handle as fh:
+        json.dump(parsed, fh)
+    os.chmod(handle.name, 0o600)
+
+    os.environ[_ADC_ENV_VAR] = handle.name
+    _written_path = handle.name
+    return handle.name
+
+
+# ============= EOF =============================================

automated_ingestion/sources/san_acacia/dlt_pipeline.py

@@ -173,6 +173,14 @@ def _approved_timestamps(
 
 def build_pipeline(environment: str) -> Any:
     """A dlt pipeline writing parquet to the raw zone for one environment."""
+    # gcsfs resolves Application Default Credentials the same way the Cloud SQL
+    # connector does, and Serverless supplies none of its own.
+    from automated_ingestion.shared.credentials import (
+        ensure_application_default_credentials,
+    )
+
+    ensure_application_default_credentials()
+
     return dlt.pipeline(
         pipeline_name=f"san_acacia_{environment}",
         destination=dlt.destinations.filesystem(

automated_ingestion/tests/test_credentials.py

@@ -0,0 +1,88 @@
+# ===============================================================================
+# Copyright 2026 ross
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+"""
+Credential materialization for a runtime with no metadata server.
+
+The failure this prevents is not subtle -- DefaultCredentialsError -- but it
+only appears in Serverless, so the tests stand in for a deployment.
+"""
+
+import json
+import os
+
+import pytest
+
+from automated_ingestion.shared import credentials
+from automated_ingestion.shared.credentials import (
+    CREDENTIALS_ENV_VAR,
+    ensure_application_default_credentials,
+)
+
+KEY = {"type": "service_account", "project_id": "waterdatainitiative-271000"}
+
+
+@pytest.fixture(autouse=True)
+def _clean(monkeypatch):
+    monkeypatch.setattr(credentials, "_written_path", None)
+    monkeypatch.delenv("GOOGLE_APPLICATION_CREDENTIALS", raising=False)
+    monkeypatch.delenv(CREDENTIALS_ENV_VAR, raising=False)
+
+
+def test_existing_credentials_are_left_alone(monkeypatch):
+    # A developer's gcloud login must not be shadowed by a key in the
+    # environment; whatever is already configured wins.
+    monkeypatch.setenv("GOOGLE_APPLICATION_CREDENTIALS", "/existing/adc.json")
+    monkeypatch.setenv(CREDENTIALS_ENV_VAR, json.dumps(KEY))
+    assert ensure_application_default_credentials() == "/existing/adc.json"
+
+
+def test_no_key_configured_is_not_an_error(monkeypatch):
+    # Locally this is normal -- google.auth finds its own credentials. In
+    # Serverless it fails later, loudly, which is the correct outcome.
+    assert ensure_application_default_credentials() is None
+    assert "GOOGLE_APPLICATION_CREDENTIALS" not in os.environ
+
+
+def test_key_is_written_and_pointed_at(monkeypatch):
+    monkeypatch.setenv(CREDENTIALS_ENV_VAR, json.dumps(KEY))
+    path = ensure_application_default_credentials()
+    assert path and os.path.exists(path)
+    assert os.environ["GOOGLE_APPLICATION_CREDENTIALS"] == path
+    with open(path) as fh:
+        assert json.load(fh) == KEY
+
+
+def test_key_file_is_not_world_readable(monkeypatch):
+    monkeypatch.setenv(CREDENTIALS_ENV_VAR, json.dumps(KEY))
+    path = ensure_application_default_credentials()
+    assert oct(os.stat(path).st_mode)[-3:] == "600"
+
+
+def test_repeated_calls_write_once(monkeypatch):
+    monkeypatch.setenv(CREDENTIALS_ENV_VAR, json.dumps(KEY))
+    first = ensure_application_default_credentials()
+    assert ensure_application_default_credentials() == first
+
+
+def test_a_path_instead_of_a_key_is_rejected(monkeypatch):
+    # Setting the variable to a filename is the obvious mistake, and it would
+    # otherwise fail much later inside google.auth.
+    monkeypatch.setenv(CREDENTIALS_ENV_VAR, "/path/to/key.json")
+    with pytest.raises(RuntimeError, match="not valid JSON"):
+        ensure_application_default_credentials()
+
+
+# ============= EOF =============================================