RT archiver v3

airflow/Dockerfile

@@ -1,4 +1,4 @@
-FROM apache/airflow:2.1.4
+FROM apache/airflow:2.1.4-python3.8
 
 # install gcloud as root, then switch back to airflow
 USER root

airflow/dags/download_gtfs_schedule_v2/download_schedule_feeds.py

@@ -21,8 +21,8 @@
 from pydantic.tools import parse_obj_as
 from requests import Session
 from typing import List, Dict
-from calitp.storage import get_fs
-from utils import (
+from calitp.storage import (
+    get_fs,
     AirtableGTFSDataExtract,
     AirtableGTFSDataRecord,
     AirtableGTFSDataRecordProcessingOutcome,
@@ -72,7 +72,7 @@ def download_feed(
         config=record,
         response_code=resp.status_code,
         response_headers=resp.headers,
-        download_time=pendulum.now(),
+        ts=pendulum.now(),
     )
 
     extract.save_content(fs=get_fs(), content=resp.content)
@@ -135,8 +135,8 @@ def download_all(task_instance, execution_date, **kwargs):
     ), f"we somehow ended up with {len(outcomes)} outcomes from {len(records)} records"
 
     result = DownloadFeedsResult(
-        start_time=start,
-        end_time=pendulum.now(),
+        ts=start,
+        end=pendulum.now(),
         outcomes=outcomes,
         filename="results.jsonl",
     )

airflow/dags/macros.py

@@ -2,7 +2,7 @@
 
 import pandas as pd
 from calitp.config import is_development
-from utils import prefix_bucket
+from calitp.storage import prefix_bucket
 
 # To add a macro, add its definition in the appropriate section
 # And then add it to the dictionary at the bottom of this file

airflow/plugins/operators/airtable_to_gcs.py

@@ -6,8 +6,7 @@
 from pyairtable import Table
 from pydantic import BaseModel
 from typing import Optional
-from utils import make_name_bq_safe
-from calitp.storage import get_fs
+from calitp.storage import get_fs, make_name_bq_safe
 
 from airflow.models import BaseOperator
 
@@ -98,7 +97,7 @@ def make_hive_path(self, bucket: str):
             bucket,
             f"{self.air_base_name}__{safe_air_table_name}",
             f"dt={self.extract_time.to_date_string()}",
-            f"time={self.extract_time.to_time_string()}",
+            f"ts={self.extract_time.to_iso8601_string()}",
             f"{safe_air_table_name}.jsonl.gz",
         )