new normalisation

Makefile

@@ -24,12 +24,12 @@ CAROOT = $(shell pwd)/infra/traefik/certs
 install:
 	@ echo -e "$(BUILD_PRINT)Installing the requirements$(END_BUILD_PRINT)"
 	@ pip install --upgrade pip
-	@ pip install --no-cache-dir -r requirements.txt --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.5.1/constraints-no-providers-3.8.txt"
+	@ pip install --no-cache-dir -r requirements.txt --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.7.1/constraints-no-providers-3.8.txt"
 
 install-dev:
 	@ echo -e "$(BUILD_PRINT)Installing the dev requirements$(END_BUILD_PRINT)"
 	@ pip install --upgrade pip
-	@ pip install --no-cache-dir -r requirements.dev.txt --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.5.1/constraints-no-providers-3.8.txt"
+	@ pip install --no-cache-dir -r requirements.dev.txt --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.7.1/constraints-no-providers-3.8.txt"
 
 test: test-unit
 
@@ -213,7 +213,7 @@ stop-metabase:
 init-rml-mapper:
 	@ echo -e "RMLMapper folder initialisation!"
 	@ mkdir -p ./.rmlmapper
-	@ wget -c https://github.com/RMLio/rmlmapper-java/releases/download/v6.1.3/rmlmapper-6.1.3-r367-all.jar -O ./.rmlmapper/rmlmapper.jar
+	@ wget -c https://github.com/RMLio/rmlmapper-java/releases/download/v6.2.1/rmlmapper-6.2.1-r368-all.jar -O ./.rmlmapper/rmlmapper.jar
 
 init-limes:
 	@ echo -e "Limes folder initialisation!"

dags/daily_check_notices_availability_in_cellar.py

@@ -1,3 +1,8 @@
+"""
+This DAG is responsible for checking the availability of notices in the cellar.
+"""
+
+
 from airflow.decorators import dag, task
 from pymongo import MongoClient
 
@@ -11,6 +16,8 @@
 
 
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
+     description=__doc__[0: __doc__.find(".")],
+     doc_md=__doc__,
      catchup=False,
      schedule_interval="0 0 * * *",
      tags=['daily', 'validation'])

dags/daily_materialized_views_update.py

@@ -1,3 +1,7 @@
+"""
+This DAG is responsible for updating the materialized views of the notice and batch collections.
+"""
+
 from airflow.decorators import dag, task
 from pymongo import MongoClient
 
@@ -13,6 +17,8 @@
 
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
      catchup=False,
+     description=__doc__[0: __doc__.find(".")],
+     doc_md=__doc__,
      schedule_interval="0 6 * * *",
      tags=['mongodb', 'daily-views-update'])
 def daily_materialized_views_update():

dags/daily_notices_metadata_update.py

@@ -0,0 +1,73 @@
+"""
+DAG to update daily notices metadata from TED.
+"""
+
+from datetime import date, datetime, timedelta
+
+from airflow.decorators import dag, task
+from airflow.models import Param
+from airflow.timetables.trigger import CronTriggerTimetable
+
+from dags import DEFAULT_DAG_ARGUMENTS
+from dags.dags_utils import get_dag_param
+from ted_sws.supra_notice_manager.services.daily_notices_metadata_services import \
+    update_daily_notices_metadata_from_ted, \
+    update_daily_notices_metadata_with_fetched_data
+
+START_DATE_PARAM_KEY = "start_date"
+END_DATE_PARAM_KEY = "end_date"
+DEFAULT_TED_API_START_DATE = "2014-01-01"
+DEFAULT_TED_API_START_DATE_FORMAT = "%Y-%m-%d"
+
+
+@dag(default_args=DEFAULT_DAG_ARGUMENTS,
+     schedule_interval=None,
+     tags=['daily', "dashboards", "metadata", "ted", "notices"],
+     catchup=False,
+     timetable=CronTriggerTimetable('0 19 * * *', timezone='UTC'),
+     description=__doc__[0: __doc__.find(".")],
+     doc_md=__doc__,
+     params={
+         START_DATE_PARAM_KEY: Param(
+             default=f"{date.today()}",
+             type="string",
+             format="date",
+             title="Start Date",
+             description="""This field is required.
+                    Start date of the date range to fetch notices from TED."""
+         ),
+         END_DATE_PARAM_KEY: Param(
+             default=f"{date.today()}",
+             type="string",
+             format="date",
+             title="End Date",
+             description="""This field is required.
+                    End date of the date range to fetch notices from TED."""
+         )
+     }
+     )
+def daily_notices_metadata_update():
+    @task
+    def update_daily_notices_metadata_from_ted_api():
+        start_date = get_dag_param(key=START_DATE_PARAM_KEY, default_value=DEFAULT_TED_API_START_DATE)
+        end_date = get_dag_param(key=END_DATE_PARAM_KEY, default_value=(datetime.today() - timedelta(days=1)).strftime(
+            DEFAULT_TED_API_START_DATE_FORMAT))
+
+        update_daily_notices_metadata_from_ted(
+            start_date=datetime.strptime(start_date, DEFAULT_TED_API_START_DATE_FORMAT),
+            end_date=datetime.strptime(end_date, DEFAULT_TED_API_START_DATE_FORMAT))
+
+    @task
+    def update_daily_notices_metadata_with_fetched_data_from_repo():
+        start_date = get_dag_param(key=START_DATE_PARAM_KEY, default_value=DEFAULT_TED_API_START_DATE)
+        end_date = get_dag_param(key=END_DATE_PARAM_KEY, default_value=(datetime.today() - timedelta(days=1)).strftime(
+            DEFAULT_TED_API_START_DATE_FORMAT))
+
+        update_daily_notices_metadata_with_fetched_data(
+            start_date=datetime.strptime(start_date, DEFAULT_TED_API_START_DATE_FORMAT),
+            end_date=datetime.strptime(end_date, DEFAULT_TED_API_START_DATE_FORMAT))
+
+    update_daily_notices_metadata_from_ted_api() >> update_daily_notices_metadata_with_fetched_data_from_repo()
+
+
+dag = daily_notices_metadata_update()

dags/fetch_notices_by_date.py

@@ -1,7 +1,12 @@
-from datetime import timedelta
+"""
+This DAG is used to fetch notices from TED by date.
+"""
+
+from datetime import date, datetime, timedelta
 
 from airflow.decorators import dag, task
-from airflow.operators.dummy import DummyOperator
+from airflow.models import Param
+from airflow.operators.empty import EmptyOperator
 from airflow.operators.python import BranchPythonOperator, PythonOperator
 from airflow.utils.trigger_rule import TriggerRule
 from airflow.timetables.trigger import CronTriggerTimetable
@@ -17,7 +22,7 @@
 
 DAG_NAME = "fetch_notices_by_date"
 BATCH_SIZE = 2000
-WILD_CARD_DAG_KEY = "wild_card"
+FETCH_DATE_DAG_KEY = "fetch_date"
 TRIGGER_COMPLETE_WORKFLOW_DAG_KEY = "trigger_complete_workflow"
 TRIGGER_PARTIAL_WORKFLOW_TASK_ID = "trigger_partial_notice_proc_workflow"
 TRIGGER_COMPLETE_WORKFLOW_TASK_ID = "trigger_complete_notice_proc_workflow"
@@ -28,8 +33,28 @@
 
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
      catchup=False,
+     description=__doc__[0: __doc__.find(".")],
+     doc_md=__doc__,
      timetable=CronTriggerTimetable('0 1 * * *', timezone='UTC'),
-     tags=['selector', 'daily-fetch'])
+     tags=['selector', 'daily-fetch'],
+     params={
+         FETCH_DATE_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."""
+         ),
+         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."""
+         )
+     }
+     )
 def fetch_notices_by_date():
     @task
     @event_log(TechnicalEventMessage(
@@ -39,7 +64,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=FETCH_DATE_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:
@@ -61,10 +90,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
-
-        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_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
+        publication_date = datetime.strptime(get_dag_param(key=FETCH_DATE_DAG_KEY,
+                                                           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)
 
@@ -91,7 +120,7 @@ def _branch_selector():
         python_callable=validate_fetched_notices
     )
 
-    finish_step = DummyOperator(task_id=FINISH_FETCH_BY_DATE_TASK_ID,
+    finish_step = EmptyOperator(task_id=FINISH_FETCH_BY_DATE_TASK_ID,
                                 trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS)
 
     fetch_by_date_notice_from_ted() >> branch_task >> [trigger_normalisation_workflow,

dags/fetch_notices_by_date_range.py

@@ -1,5 +1,11 @@
-from datetime import datetime
+"""
+This DAG is responsible for fetching notices from TED by date range.
+"""
+
+from datetime import datetime, date
 from typing import Any
+
+from airflow.models import Param
 from dateutil import rrule
 
 from airflow.decorators import dag, task
@@ -8,7 +14,7 @@
 
 from dags import DEFAULT_DAG_ARGUMENTS
 from dags.dags_utils import get_dag_param
-from dags.fetch_notices_by_date import WILD_CARD_DAG_KEY, TRIGGER_COMPLETE_WORKFLOW_DAG_KEY, \
+from dags.fetch_notices_by_date import FETCH_DATE_DAG_KEY, TRIGGER_COMPLETE_WORKFLOW_DAG_KEY, \
     DAG_NAME as FETCH_NOTICES_BY_DATE_DAG_NAME
 from ted_sws.event_manager.adapters.event_log_decorator import event_log
 from ted_sws.event_manager.model.event_message import TechnicalEventMessage, EventMessageMetadata, \
@@ -20,20 +26,48 @@
 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=['master'])
+@dag(default_args=DEFAULT_DAG_ARGUMENTS, schedule_interval=None, tags=['master'],
+     description=__doc__[0: __doc__.find(".")],
+     doc_md=__doc__,
+     params={
+         START_DATE_KEY: Param(
+             default=f"{date.today()}",
+             type="string",
+             format="date",
+             title="Start Date",
+             description="""This field is required.
+                    Start date of the date range to fetch notices from TED."""
+         ),
+         END_DATE_KEY: Param(
+             default=f"{date.today()}",
+             type="string",
+             format="date",
+             title="End Date",
+             description="""This field is required.
+                    End date of the date range to fetch notices from TED."""
+         ),
+         TRIGGER_COMPLETE_WORKFLOW_DAG_KEY: Param(
+             default=False,
+             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."""
+         )
+     }
+     )
 def fetch_notices_by_date_range():
     @task
     @event_log(TechnicalEventMessage(
@@ -46,13 +80,14 @@ def trigger_notice_by_date_for_each_date_in_range():
         context: Any = get_current_context()
         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:
+        trigger_complete_workflow = get_dag_param(key=TRIGGER_COMPLETE_WORKFLOW_DAG_KEY,
+                                                  default_value=False)
+        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={FETCH_DATE_DAG_KEY: fetch_date.strftime('%Y-%m-%d'),
                       TRIGGER_COMPLETE_WORKFLOW_DAG_KEY: trigger_complete_workflow,
                       }
             ).execute(context=context)

dags/fetch_notices_by_query.py

@@ -1,5 +1,12 @@
+"""
+This DAG is responsible for fetching notices from TED by query.
+"""
+
+from datetime import date
+
 from airflow.decorators import dag, task
-from airflow.operators.dummy import DummyOperator
+from airflow.models import Param
+from airflow.operators.empty import EmptyOperator
 from airflow.operators.python import BranchPythonOperator
 from airflow.utils.trigger_rule import TriggerRule
 from dags import DEFAULT_DAG_ARGUMENTS
@@ -22,7 +29,26 @@
 
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
      schedule_interval=None,
-     tags=['fetch'])
+     tags=['fetch'],
+     description=__doc__[0: __doc__.find(".")],
+     doc_md=__doc__,
+     params={
+         QUERY_DAG_KEY: Param(
+             default=f"PD=[>={date.today().strftime('%Y%m%d')} AND <={date.today().strftime('%Y%m%d')}]",
+             type="string",
+             title="Query",
+             description="""This field is required.
+                Query 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."""
+         )
+     }
+     )
 def fetch_notices_by_query():
     @task
     @event_log(TechnicalEventMessage(
@@ -58,7 +84,7 @@ def _branch_selector():
         python_callable=_branch_selector,
     )
 
-    finish_step = DummyOperator(task_id=FINISH_FETCH_BY_DATE_TASK_ID,
+    finish_step = EmptyOperator(task_id=FINISH_FETCH_BY_DATE_TASK_ID,
                                 trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS)
 
     fetch_by_query_notice_from_ted() >> branch_task >> [trigger_normalisation_workflow,