fixes for fetch by dates

Author: Dragos0000

dags/fetch_notices_by_date.py

@@ -1,4 +1,4 @@
-from datetime import timedelta, date
+from datetime import timedelta, date, datetime
 
 from airflow.decorators import dag, task
 from airflow.models import Param
@@ -37,18 +37,19 @@
      tags=['selector', 'daily-fetch'],
      params={
          WILD_CARD_DAG_KEY: Param(
+             default=f"{date.today() - timedelta(days=1)}",
              type="string",
              format="date",
              title="Date",
              description="""This field is required.
-                          Date to fetch notices from TED. Correct format is YYYYMMDD*"""
+                            Date to fetch notices from TED."""
          ),
          TRIGGER_COMPLETE_WORKFLOW_DAG_KEY: Param(
              default=True,
              type="boolean",
              title="Trigger Complete Workflow",
              description="""This field is required.
-                            If true, the complete workflow will be triggered, otherwise only the partial workflow will be triggered."""
+                              If true, the complete workflow will be triggered, otherwise only the partial workflow will be triggered."""
          )
      }
      )
@@ -61,7 +62,11 @@ def fetch_notices_by_date():
         ))
     )
     def fetch_by_date_notice_from_ted():
-        notice_ids = notice_fetcher_by_date_pipeline(date_wild_card=get_dag_param(key=WILD_CARD_DAG_KEY))
+        default_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
+        selected_date = datetime.strptime(get_dag_param(key=WILD_CARD_DAG_KEY,
+                                                        default_value=default_date), "%Y-%m-%d")
+        date_wild_card = datetime.strftime(selected_date, "%Y%m%d*")
+        notice_ids = notice_fetcher_by_date_pipeline(date_wild_card=date_wild_card)
         if not notice_ids:
             log_error("No notices has been fetched!")
         else:
@@ -83,10 +88,10 @@ def validate_fetched_notices():
         from ted_sws.supra_notice_manager.services.supra_notice_validator import validate_and_update_daily_supra_notice
         from datetime import datetime
         from pymongo import MongoClient
-
+        default_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
         publication_date = datetime.strptime(get_dag_param(key=WILD_CARD_DAG_KEY,
-                                                           default_value=(datetime.now() - timedelta(days=1)).strftime(
-                                                               "%Y%m%d*")), "%Y%m%d*")
+                                                           default_value=default_date
+                                                           ), "%Y-%m-%d")
         mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
         validate_and_update_daily_supra_notice(ted_publication_date=publication_date, mongodb_client=mongodb_client)
 

dags/fetch_notices_by_date_range.py

@@ -21,20 +21,20 @@
 END_DATE_KEY = "end_date"
 
 
-def generate_wildcards_foreach_day_in_range(start_date: str, end_date: str) -> list:
+def generate_list_of_dates_from_date_range(start_date: str, end_date: str) -> list:
     """
         Given a date range returns all daily dates in that range
     :param start_date:
     :param end_date:
     :return:
     """
-    return [dt.strftime('%Y%m%d*')
+    return [dt
             for dt in rrule.rrule(rrule.DAILY,
-                                  dtstart=datetime.strptime(start_date, '%Y%m%d'),
-                                  until=datetime.strptime(end_date, '%Y%m%d'))]
+                                  dtstart=datetime.strptime(start_date, '%Y-%m-%d'),
+                                  until=datetime.strptime(end_date, '%Y-%m-%d'))]
 
 
-@dag(default_args=DEFAULT_DAG_ARGUMENTS, schedule_interval=None, tags=['fetch'],
+@dag(default_args=DEFAULT_DAG_ARGUMENTS, schedule_interval=None, tags=['master'],
      params={
          START_DATE_KEY: Param(
              default=f"{date.today()}",
@@ -74,12 +74,12 @@ def trigger_notice_by_date_for_each_date_in_range():
         start_date = get_dag_param(key=START_DATE_KEY, raise_error=True)
         end_date = get_dag_param(key=END_DATE_KEY, raise_error=True)
         trigger_complete_workflow = get_dag_param(key=TRIGGER_COMPLETE_WORKFLOW_DAG_KEY, default_value=False)
-        date_wildcards = generate_wildcards_foreach_day_in_range(start_date, end_date)
-        for date_wildcard in date_wildcards:
+        fetch_dates = generate_list_of_dates_from_date_range(start_date, end_date)
+        for fetch_date in fetch_dates:
             TriggerDagRunOperator(
-                task_id=f'trigger_notice_fetch_by_date_workflow_dag_{date_wildcard[:-1]}',
+                task_id=f'trigger_notice_fetch_by_date_workflow_dag_{fetch_date.strftime("%Y_%m_%d")}',
                 trigger_dag_id=FETCH_NOTICES_BY_DATE_DAG_NAME,
-                conf={WILD_CARD_DAG_KEY: date_wildcard,
+                conf={WILD_CARD_DAG_KEY: fetch_date.strftime('%Y-%m-%d'),
                       TRIGGER_COMPLETE_WORKFLOW_DAG_KEY: trigger_complete_workflow,
                       }
             ).execute(context=context)

infra/airflow/docker-compose.yaml

@@ -60,6 +60,7 @@ x-airflow-common:
     AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
     AIRFLOW__CORE__ENABLE_XCOM_PICKLING: "true"
     AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
+#    AIRFLOW__WEBSERVER__SHOW_TRIGGER_FORM_IF_NO_PARAMS: 'true'
     AIRFLOW__API__AUTH_BACKEND: 'airflow.api.auth.backend.basic_auth'
     _PIP_ADDITIONAL_REQUIREMENTS: ${_PIP_ADDITIONAL_REQUIREMENTS:-}
     VAULT_TOKEN: ${VAULT_TOKEN}
@@ -90,7 +91,7 @@ x-airflow-common:
 
 services:
   postgres:
-    image: postgres:13
+    image: postgres:14.4-alpine
     container_name: postgres-airflow-${ENVIRONMENT}
     environment:
       POSTGRES_USER: ${AIRFLOW_POSTGRES_USER}