adjust max runs and tasks; bump retries by 1

Author: adamscarberry

dags/cumulus/aprfc_qte_01h.py

@@ -20,11 +20,13 @@
 default_args = {
     "owner": "airflow",
     "depends_on_past": False,
-    "start_date": (datetime.utcnow() - timedelta(hours=48)).replace(minute=0, second=0, microsecond=0),
+    "start_date": (datetime.utcnow() - timedelta(hours=48)).replace(
+        minute=0, second=0, microsecond=0
+    ),
     "catchup_by_default": False,
     "email_on_failure": False,
     "email_on_retry": False,
-    "retries": 5,
+    "retries": 6,
     "retry_delay": timedelta(minutes=30),
 }
 
@@ -33,8 +35,8 @@
     default_args=default_args,
     tags=["cumulus", "AIRTEMP", "QTE", "APRFC"],
     schedule="45 * * * *",
-    max_active_runs=2,
-    max_active_tasks=4,
+    max_active_runs=1,
+    max_active_tasks=1,
 )
 def cumulus_aprfc_qte_01h():
     """
@@ -54,47 +56,36 @@ def cumulus_aprfc_qte_01h():
     s3_bucket = cumulus.S3_BUCKET
     key_prefix = cumulus.S3_ACQUIRABLE_PREFIX
 
-    
-
     URL_ROOT = "https://nomads.ncep.noaa.gov/pub/data/nccf/com/urma/prod/"
     PRODUCT_SLUG = "aprfc-qte-01h"
 
-    filename_template = Template(
-        "akurma.t${hr_}z.2dvaranl_ndfd_3p0.grb2 "
-    )
-
-    url_suffix_template = Template(
-        "akurma.${date_}"
-    )
+    filename_template = Template("akurma.t${hr_}z.2dvaranl_ndfd_3p0.grb2 ")
 
+    url_suffix_template = Template("akurma.${date_}")
 
     @task()
     def download_raw_qte():
         logical_date = get_current_context()["logical_date"]
         date_only = logical_date.strftime("%Y%m%d")
 
         url_suffix = url_suffix_template.substitute(
-                        date_=date_only,
-                    )
+            date_=date_only,
+        )
 
         filename = filename_template.substitute(
-                    hr_=logical_date.strftime("%H"),
-                    )
+            hr_=logical_date.strftime("%H"),
+        )
 
         file_dir = f"{URL_ROOT}{url_suffix}"
 
-
-        s3_filename = f'{date_only}_{filename}'
+        s3_filename = f"{date_only}_{filename}"
         s3_key = f"{key_prefix}/{PRODUCT_SLUG}/{s3_filename}"
 
-
-        
-
         print(f"Downloading file: {filename}")
 
         trigger_download(
-                    url=f"{file_dir}/{filename}", s3_bucket=s3_bucket, s3_key=s3_key
-                )
+            url=f"{file_dir}/{filename}", s3_bucket=s3_bucket, s3_key=s3_key
+        )
         return json.dumps(
             {
                 "execution": logical_date.isoformat(),
@@ -103,11 +94,6 @@ def download_raw_qte():
             }
         )
 
-
-
-
-
-
     @task()
     def notify_cumulus(payload):
         payload = json.loads(payload)
@@ -116,11 +102,9 @@ def notify_cumulus(payload):
             acquirable_id=cumulus.acquirables[PRODUCT_SLUG],
             datetime=payload["execution"],
             s3_key=payload["s3_key"],
-            )
+        )
 
     notify_cumulus(download_raw_qte())
-    
-
 
 
 aprfc_qte_dag = cumulus_aprfc_qte_01h()