Merge pull request #310 from cityofaustin/charlie/311-reporting-updates

Open311 Scraping DAG

Author: Charlie-Henry

dags/dts_csr_report_publishing.py dags/dts_311_report_publishing.pydags/dts_csr_report_publishing.py renamed to dags/dts_311_report_publishing.py

@@ -1,4 +1,4 @@
-# test locally with: docker compose run --rm airflow-cli dags test dts_csr_report_publishing
+# test locally with: docker compose run --rm airflow-cli dags test dts_311_report_publishing
 
 from os import getenv
 
@@ -49,7 +49,7 @@
         "opitem": "Executive Dashboard",
         "opfield": "datasets.Revenue",
     },
-    "CSR_DATASET": {
+    "REQUESTS_DATASET": {
         "opitem": "Executive Dashboard",
         "opfield": "datasets.CSR",
     },
@@ -92,7 +92,7 @@
 }
 
 CUR_YEAR_SECRETS = {
-    "CSR_ENDPOINT": {
+    "REQUESTS_ENDPOINT": {
         "opitem": "Executive Dashboard",
         "opfield": "csr.Current FY Endpoint",
     },
@@ -107,7 +107,7 @@
 }
 
 PREV_YEAR_SECRETS = {
-    "CSR_ENDPOINT": {
+    "REQUESTS_ENDPOINT": {
         "opitem": "Executive Dashboard",
         "opfield": "csr.Previous FY Endpoint",
     },
@@ -122,7 +122,7 @@
 }
 
 TWO_YEARS_AGO_SECRETS = {
-    "CSR_ENDPOINT": {
+    "REQUESTS_ENDPOINT": {
         "opitem": "Executive Dashboard",
         "opfield": "csr.Two Years Ago FY Endpoint",
     },
@@ -142,14 +142,14 @@
 TWO_YEARS_AGO_SECRETS.update(OTHER_SECRETS)
 
 with DAG(
-    dag_id="dts_csr_report_publishing",
+    dag_id="dts_311_report_publishing",
     description="Downloads reports of 311 service requests for TPW and publishes it in a Socrata dataset.",
     default_args=default_args,
     schedule_interval=(
         "36 2,13 * * *" if DEPLOYMENT_ENVIRONMENT == "production" else None
     ),
     dagrun_timeout=timedelta(minutes=60),
-    tags=["repo:dts-311-reporting", "socrata", "csr"],
+    tags=["repo:dts-311-reporting", "socrata", "311"],
     catchup=False,
 ) as dag:
     docker_image = "atddocker/dts-311-reporting:production"
@@ -159,12 +159,12 @@
     two_years_env = get_env_vars_task(TWO_YEARS_AGO_SECRETS)
 
     t1 = DockerOperator(
-        task_id="cur_year_csr_report_to_socrata",
+        task_id="cur_year_requests_report_to_socrata",
         image=docker_image,
         docker_conn_id="docker_default",
         api_version="auto",
         auto_remove="force",
-        command=f"python etl/csr_to_socrata.py",
+        command="python -m etl.csv_reporting.requests_to_socrata",
         environment=cur_year_env,
         tty=True,
         force_pull=True,
@@ -176,7 +176,7 @@
         docker_conn_id="docker_default",
         api_version="auto",
         auto_remove="force",
-        command=f"python etl/flex_notes_to_socrata.py",
+        command="python -m etl.csv_reporting.flex_notes_to_socrata",
         environment=cur_year_env,
         tty=True,
     )
@@ -187,18 +187,18 @@
         docker_conn_id="docker_default",
         api_version="auto",
         auto_remove="force",
-        command=f"python etl/activities_to_socrata.py",
+        command="python -m etl.csv_reporting.activities_to_socrata",
         environment=cur_year_env,
         tty=True,
     )
 
     t4 = DockerOperator(
-        task_id="prev_year_csr_report_to_socrata",
+        task_id="prev_year_requests_report_to_socrata",
         image=docker_image,
         docker_conn_id="docker_default",
         api_version="auto",
         auto_remove="force",
-        command=f"python etl/csr_to_socrata.py",
+        command="python -m etl.csv_reporting.requests_to_socrata",
         environment=prev_year_env,
         tty=True,
     )
@@ -209,7 +209,7 @@
         docker_conn_id="docker_default",
         api_version="auto",
         auto_remove="force",
-        command=f"python etl/flex_notes_to_socrata.py",
+        command="python -m etl.csv_reporting.flex_notes_to_socrata",
         environment=prev_year_env,
         tty=True,
     )
@@ -220,18 +220,18 @@
         docker_conn_id="docker_default",
         api_version="auto",
         auto_remove="force",
-        command=f"python etl/activities_to_socrata.py",
+        command="python -m etl.csv_reporting.activities_to_socrata",
         environment=prev_year_env,
         tty=True,
     )
 
     t7 = DockerOperator(
-        task_id="two_years_ago_csr_report_to_socrata",
+        task_id="two_years_ago_requests_report_to_socrata",
         image=docker_image,
         docker_conn_id="docker_default",
         api_version="auto",
         auto_remove="force",
-        command=f"python etl/csr_to_socrata.py",
+        command="python -m etl.csv_reporting.requests_to_socrata",
         environment=two_years_env,
         tty=True,
     )
@@ -242,7 +242,7 @@
         docker_conn_id="docker_default",
         api_version="auto",
         auto_remove="force",
-        command=f"python etl/flex_notes_to_socrata.py",
+        command="python -m etl.csv_reporting.flex_notes_to_socrata",
         environment=two_years_env,
         tty=True,
     )
@@ -253,7 +253,7 @@
         docker_conn_id="docker_default",
         api_version="auto",
         auto_remove="force",
-        command=f"python etl/activities_to_socrata.py",
+        command="python -m etl.csv_reporting.activities_to_socrata",
         environment=two_years_env,
         tty=True,
     )

dags/dts_open_311_scrape.py

@@ -0,0 +1,89 @@
+# test locally with: docker compose run --rm airflow-cli dags test dts_open_311_scrape
+
+from os import getenv
+
+from datetime import timedelta
+
+from airflow.decorators import task
+from airflow.models import DAG
+from airflow.operators.docker_operator import DockerOperator
+from pendulum import datetime, duration, now
+
+from utils.onepassword import get_env_vars_task
+from utils.slack_operator import task_fail_slack_alert
+from utils.time import get_previous_run_date
+
+DEPLOYMENT_ENVIRONMENT = getenv("ENVIRONMENT", "development")
+
+default_args = {
+    "owner": "airflow",
+    "depends_on_past": False,
+    "start_date": datetime(2015, 12, 1, tz="America/Chicago"),
+    "email_on_failure": False,
+    "email_on_retry": False,
+    "retries": 0,
+    "on_failure_callback": task_fail_slack_alert,
+}
+
+REQUIRED_SECRETS = {
+    "SO_WEB": {
+        "opitem": "Socrata Key ID, Secret, and Token",
+        "opfield": "socrata.endpoint",
+    },
+    "SO_TOKEN": {
+        "opitem": "Socrata Key ID, Secret, and Token",
+        "opfield": "socrata.appToken",
+    },
+    "SO_SECRET": {
+        "opitem": "Socrata Key ID, Secret, and Token",
+        "opfield": "socrata.apiKeySecret",
+    },
+    "SO_KEY": {
+        "opitem": "Socrata Key ID, Secret, and Token",
+        "opfield": "socrata.apiKeyId",
+    },
+    "REALTIME_DATASET": {
+        "opitem": "Executive Dashboard",
+        "opfield": "datasets.Realtime",
+    },
+    "OPEN_311_API_KEY": {
+        "opitem": "Austin Open 311 API key",
+        "opfield": "production.API key",
+    },
+    "OPEN_311_API_BASE_URL": {
+        "opitem": "Austin Open 311 API key",
+        "opfield": "production.Base URL",
+    },
+}
+
+
+with DAG(
+    dag_id="dts_open_311_scrape",
+    description="Downloads CSRs from Open311 API and publishes them in a Socrata dataset",
+    default_args=default_args,
+    schedule_interval=(
+        "*/5 * * * *" if DEPLOYMENT_ENVIRONMENT == "production" else None
+    ),
+    dagrun_timeout=timedelta(minutes=5),
+    tags=["repo:dts-311-reporting", "socrata", "311", "open311"],
+    catchup=False,
+) as dag:
+    docker_image = "atddocker/dts-311-reporting:production"
+
+    env_vars = get_env_vars_task(REQUIRED_SECRETS)
+    one_day_ago = now("America/Chicago").subtract(days=1)
+    date_arg = get_previous_run_date(fallback_date=one_day_ago.to_iso8601_string())
+
+    t1 = DockerOperator(
+        task_id="open311_to_socrata",
+        image=docker_image,
+        docker_conn_id="docker_default",
+        api_version="auto",
+        auto_remove="force",
+        command=f"python -m etl.open311.open311_to_socrata -d {date_arg["last_run_datetime_iso"]}",
+        environment=env_vars,
+        tty=True,
+        force_pull=True,
+    )
+
+    t1

dags/utils/time.py

@@ -6,7 +6,7 @@
     task_id="get_previous_run_date",
     multiple_outputs=True,
 )
-def get_previous_run_date(**context):
+def get_previous_run_date(fallback_date="1970-01-01", **context):
     """Task to return the last successful run date in UTC datetime format.
 
     Args:
@@ -16,9 +16,9 @@ def get_previous_run_date(**context):
     Returns:
         Dict: dict containing the last run datetime and other future formats
     """
-    last_run_datetime = context.get("prev_start_date_success") or parse("1970-01-01")
+    last_run_datetime = context.get("prev_start_date_success") or parse(fallback_date)
 
     return {
         "last_run_datetime": last_run_datetime,
-        # add other formats here
+        "last_run_datetime_iso": last_run_datetime.isoformat(),
     }