Increases timeout for data extraction job to prevent errors. Updates data transformation job to use date associated with data extraction job. Fixes configuration of boto3 and smart_open clients to permit file writes to Cloudflare R2.

LaunaG · LaunaG · commit cb45ad5bcc16 · 2025-11-06T00:16:00.000-06:00
diff --git a/.github/workflows/deploy-gcp.yaml b/.github/workflows/deploy-gcp.yaml
@@ -112,6 +112,7 @@ jobs:
         env:
           CLOUDFLARE_R2_ACCESS_KEY_ID: ${{ secrets.CLOUDFLARE_R2_ACCESS_KEY_ID }}
           CLOUDFLARE_R2_BUCKET_URL: ${{ vars.CLOUDFLARE_R2_BUCKET_URL }}
+          CLOUDFLARE_R2_ENDPOINT_URL: ${{ vars.CLOUDFLARE_R2_ENDPOINT_URL }}
           CLOUDFLARE_R2_SECRET_ACCESS_KEY: ${{ secrets.CLOUDFLARE_R2_SECRET_ACCESS_KEY }}
           DJANGO_PORT: ${{ vars.DJANGO_PORT }}
           DJANGO_SECRET_KEY: ${{ secrets.DJANGO_SECRET_KEY }}
@@ -125,6 +126,7 @@ jobs:
           GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           OUTPUT_FILE_MAX_AGE: ${{ vars.OUTPUT_FILE_MAX_AGE }}
+          OUTPUT_FILE_NAME: ${{ vars.OUTPUT_FILE_NAME }}
           OUTPUT_FILE_TOTAL_MAX_ATTEMPTS: ${{ vars.OUTPUT_FILE_TOTAL_MAX_ATTEMPTS }}
           POSTGRES_DB: ${{ vars.POSTGRES_DB }}
           POSTGRES_PASSWORD: ${{ secrets.POSTGRES_PASSWORD }}
@@ -142,6 +144,7 @@ jobs:
         env:
           CLOUDFLARE_R2_ACCESS_KEY_ID: ${{ secrets.CLOUDFLARE_R2_ACCESS_KEY_ID }}
           CLOUDFLARE_R2_BUCKET_URL: ${{ vars.CLOUDFLARE_R2_BUCKET_URL }}
+          CLOUDFLARE_R2_ENDPOINT_URL: ${{ vars.CLOUDFLARE_R2_ENDPOINT_URL }}
           CLOUDFLARE_R2_SECRET_ACCESS_KEY: ${{ secrets.CLOUDFLARE_R2_SECRET_ACCESS_KEY }}
           DJANGO_PORT: ${{ vars.DJANGO_PORT }}
           DJANGO_SECRET_KEY: ${{ secrets.DJANGO_SECRET_KEY }}
@@ -155,6 +158,7 @@ jobs:
           GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           OUTPUT_FILE_MAX_AGE: ${{ vars.OUTPUT_FILE_MAX_AGE }}
+          OUTPUT_FILE_NAME: ${{ vars.OUTPUT_FILE_NAME }}
           OUTPUT_FILE_TOTAL_MAX_ATTEMPTS: ${{ vars.OUTPUT_FILE_TOTAL_MAX_ATTEMPTS }}
           POSTGRES_DB: ${{ vars.POSTGRES_DB }}
           POSTGRES_PASSWORD: ${{ secrets.POSTGRES_PASSWORD }}
diff --git a/infra/gcp/persistent/__main__.py b/infra/gcp/persistent/__main__.py
@@ -10,6 +10,7 @@
 from constants import (
     CLOUDFLARE_R2_ACCESS_KEY_ID,
     CLOUDFLARE_R2_BUCKET_URL,
+    CLOUDFLARE_R2_ENDPOINT_URL,
     CLOUDFLARE_R2_SECRET_ACCESS_KEY,
     DJANGO_ALLOWED_HOST,
     DJANGO_API_PATH_DATA_EXTRACTION,
@@ -26,6 +27,7 @@
     IS_TEST,
     MAPPING_DIR,
     OUTPUT_FILE_MAX_AGE,
+    OUTPUT_FILE_NAME,
     OUTPUT_FILE_TOTAL_MAX_ATTEMPTS,
     POSTGRES_DB,
     POSTGRES_PASSWORD,
@@ -982,7 +984,7 @@
                 )
             ],
             service_account=cloud_run_service_account.email,
-            timeout="86400s",
+            timeout="172800s",
             volumes=[
                 gcp.cloudrunv2.JobTemplateTemplateVolumeArgs(
                     name="cloudsql",
@@ -1149,6 +1151,10 @@
                             name="CLOUDFLARE_R2_ACCESS_KEY_ID",
                             value=CLOUDFLARE_R2_ACCESS_KEY_ID,
                         ),
+                        gcp.cloudrunv2.JobTemplateTemplateContainerEnvArgs(
+                            name="CLOUDFLARE_R2_ENDPOINT_URL",
+                            value=CLOUDFLARE_R2_ENDPOINT_URL,
+                        ),
                         gcp.cloudrunv2.JobTemplateTemplateContainerEnvArgs(
                             name="CLOUDFLARE_R2_SECRET_ACCESS_KEY",
                             value=CLOUDFLARE_R2_SECRET_ACCESS_KEY,
@@ -1551,8 +1557,9 @@
                 - initializeVariables:
                     assign:
                         - inputBucket: ${{ "gs://" + event.data.bucket }}
-                        - outputBucket: {output_bucket_url}
-                        - objectKey: ${{ event.data.name }}
+                        - inputObjectKey: ${{ event.data.name }}
+                        - outputBucket: {output_bucket}
+                        - outputObjectKey: {output_object_key}
                         - jobPrefix: projects/{project_id}/locations/{project_region}/jobs/
                         - mapJobFullName: ${{ jobPrefix + "{map_job_name}" }}
                 - mapData:
@@ -1563,7 +1570,8 @@
                             overrides:
                                 containerOverrides:
                                     args:
-                                        - ${{ objectKey }}
+                                        - ${{ inputObjectKey }}
+                                        - ${{ outputObjectKey }}
                                         - --input_bucket
                                         - ${{ inputBucket }}
                                         - --output_bucket
@@ -1574,9 +1582,10 @@
                     result: mapDataOperation
         """,
         map_job_name=map_cloud_run_job.name,
+        output_bucket=CLOUDFLARE_R2_BUCKET_URL,
+        output_object_key=OUTPUT_FILE_NAME,
         project_id=PROJECT_ID,
         project_region=PROJECT_REGION,
-        output_bucket_url=CLOUDFLARE_R2_BUCKET_URL,
     ),
     opts=pulumi.ResourceOptions(
         depends_on=enabled_services, provider=gcp_provider
diff --git a/infra/gcp/persistent/constants.py b/infra/gcp/persistent/constants.py
@@ -107,6 +107,7 @@
 try:
     CLOUDFLARE_R2_ACCESS_KEY_ID = os.environ["CLOUDFLARE_R2_ACCESS_KEY_ID"]
     CLOUDFLARE_R2_BUCKET_URL = os.environ["CLOUDFLARE_R2_BUCKET_URL"]
+    CLOUDFLARE_R2_ENDPOINT_URL = os.environ["CLOUDFLARE_R2_ENDPOINT_URL"]
     CLOUDFLARE_R2_SECRET_ACCESS_KEY = os.environ[
         "CLOUDFLARE_R2_SECRET_ACCESS_KEY"
     ]
@@ -123,6 +124,7 @@
     EXTRACTION_PIPELINE_SCHEDULE = os.environ["EXTRACTION_PIPELINE_SCHEDULE"]
     GEMINI_API_KEY = os.environ["GEMINI_API_KEY"]
     OUTPUT_FILE_MAX_AGE = os.environ["OUTPUT_FILE_MAX_AGE"]
+    OUTPUT_FILE_NAME = os.environ["OUTPUT_FILE_NAME"]
     OUTPUT_FILE_TOTAL_MAX_ATTEMPTS = os.environ[
         "OUTPUT_FILE_TOTAL_MAX_ATTEMPTS"
     ]
diff --git a/services/map/Makefile b/services/map/Makefile
@@ -15,6 +15,7 @@ container_output_dir := "app/output"
 project_name := $(current_dir)
 env_path := $(current_abs_path).env.dev
 input_file := cleaned.parquet
+output_file := debit_projects.parquet
 
 # Define commands
 build-mapper:
@@ -27,4 +28,4 @@ run-mapper: build-mapper
 		-v $(input_dir):/$(container_input_dir) \
 		-v $(output_dir):/$(container_output_dir) \
 		--env-file $(env_path) \
-		--rm $(project_name) ${input_file}
+		--rm $(project_name) ${input_file} ${output_file}
diff --git a/services/map/src/map_clean/constants.py b/services/map/src/map_clean/constants.py
@@ -7,4 +7,3 @@
 RUNTIME_DIR = Path.cwd()
 INPUT_DIR = RUNTIME_DIR / "input"
 OUTPUT_DIR = RUNTIME_DIR / "output"
-OUTPUT_FNAME = "debit_projects.parquet"
diff --git a/services/map/src/map_clean/main.py b/services/map/src/map_clean/main.py
@@ -11,7 +11,7 @@
 import smart_open
 
 # Package imports
-from map_clean.constants import INPUT_DIR, OUTPUT_DIR, OUTPUT_FNAME
+from map_clean.constants import INPUT_DIR, OUTPUT_DIR
 from map_clean.utils import configure_cloudflare_request_params, LoggerFactory
 
 
@@ -156,10 +156,15 @@ def get_document(row: pd.Series) -> str:
     # Parse command line arguments
     parser = argparse.ArgumentParser()
     parser.add_argument(
-        "object_key",
+        "input_object_key",
         type=str,
         help="The path to the input file in the storage directory or bucket.",
     )
+    parser.add_argument(
+        "output_object_key",
+        type=str,
+        help="The path to the output file in the storage directory or bucket.",
+    )
     parser.add_argument(
         "--input_bucket",
         type=str,
@@ -179,14 +184,21 @@ def get_document(row: pd.Series) -> str:
     )
     args = parser.parse_args()
 
-    # Validate object key argument received
-    if not args.object_key:
+    # Validate positional arguments received
+    if not args.input_object_key:
         logger.error(
             "Missing positional argument for the path to the "
             "input file in the storage directory or bucket."
         )
         exit(1)
 
+    if not args.output_object_key:
+        logger.error(
+            "Missing positional argument for the path to the "
+            "output file in the storage directory or bucket."
+        )
+        exit(1)
+
     # Determine path for input and output files
     if args.remote:
         # Validate bucket options present if files hosted remotely
@@ -198,10 +210,10 @@ def get_document(row: pd.Series) -> str:
             exit(1)
 
         # Compose path to remote input file
-        input_fpath = f"{args.input_bucket}/{args.object_key}"
+        input_fpath = f"{args.input_bucket}/{args.input_object_key}"
 
         # Compose path to remote output file
-        output_fpath = f"{args.output_bucket}/{OUTPUT_FNAME}"
+        output_fpath = f"{args.output_bucket}/{args.output_object_key}"
 
         # Compose storage transport parameters
         try:
@@ -211,11 +223,11 @@ def get_document(row: pd.Series) -> str:
             exit(1)
     else:
         # Compose path to local input file
-        input_fpath = f"{INPUT_DIR}/{args.object_key}"
+        input_fpath = f"{INPUT_DIR}/{args.input_object_key}"
 
         # Compose path to local output file
         Path.mkdir(OUTPUT_DIR, exist_ok=True)
-        output_fpath = f"{OUTPUT_DIR}/{OUTPUT_FNAME}"
+        output_fpath = f"{OUTPUT_DIR}/{args.output_object_key}"
 
         # Compose storage transport parameters
         transport_params = {}
diff --git a/services/map/src/map_clean/utils/storage.py b/services/map/src/map_clean/utils/storage.py
@@ -28,6 +28,7 @@ def configure_cloudflare_request_params() -> dict:
     # Parse environment variables
     try:
         cloudflare_access_key_id = os.environ["CLOUDFLARE_R2_ACCESS_KEY_ID"]
+        cloudflare_r2_endpoint_url = os.environ["CLOUDFLARE_R2_ENDPOINT_URL"]
         cloudflare_secret_access_key = os.environ["CLOUDFLARE_R2_SECRET_ACCESS_KEY"]
         output_file_max_age = int(os.getenv("OUTPUT_FILE_MAX_AGE", "3600"))
         output_file_max_attempts = int(os.getenv("OUTPUT_FILE_TOTAL_MAX_ATTEMPTS", "3"))
@@ -53,7 +54,11 @@ def configure_cloudflare_request_params() -> dict:
             aws_access_key_id=cloudflare_access_key_id,
             aws_secret_access_key=cloudflare_secret_access_key,
         )
-        client = session.client("s3", config=config)
+        client = session.client(
+            "s3",
+            endpoint_url=cloudflare_r2_endpoint_url,
+            config=config,
+        )
     except Exception as e:
         raise RuntimeError(
             "Failed to configure new client for writing "
diff --git a/services/transform/src/clean_raw/main.py b/services/transform/src/clean_raw/main.py
@@ -3,7 +3,6 @@
 # Standard library imports
 import argparse
 import logging
-from datetime import datetime, UTC
 from pathlib import Path
 
 # Third-party imports
@@ -206,10 +205,13 @@ def main(input_fpath: str, output_fpath: str, logger: logging.Logger) -> None:
         # Compose path to remote input file
         input_fpath = f"{args.input_bucket}/{args.object_key}"
 
+        # Parse input file name and date from input file object key
+        input_fpath_segments = args.object_key.split("/")
+        input_fname = input_fpath_segments[-1]
+        input_date = input_fpath_segments[-2]
+
         # Compose path to remote output file
-        today = datetime.now(tz=UTC).strftime("%Y-%m-%d")
-        input_fname = args.object_key.split("/")[-1]
-        output_obj_key = f"transformation/{today}/{input_fname}.parquet"
+        output_obj_key = f"transformation/{input_date}/{input_fname}.parquet"
         output_fpath = f"{args.output_bucket}/{output_obj_key}"
     else:
         # Compose path to local input file
