Feature/ted 25

dags/__init__.py

@@ -0,0 +1,16 @@
+from datetime import datetime, timedelta
+
+DEFAULT_DAG_ARGUMENTS = {
+    "owner": "airflow",
+    "depends_on_past": False,
+    "start_date": datetime.now(),
+    "email": ["info@meaningfy.ws"],
+    "email_on_failure": False,
+    "email_on_retry": False,
+    "retries": 0,
+    "retry_delay": timedelta(minutes=3600),
+    "schedule_interval": "@once",
+    "max_active_runs": 128,
+    "concurrency": 128,
+    "execution_timeout": timedelta(hours=24),
+}

dags/dags_utils.py

@@ -0,0 +1,36 @@
+from airflow.operators.python import get_current_context
+
+TASK_INSTANCE = "ti"
+
+
+def select_first_non_none(data):
+    """
+
+    :param data:
+    :return:
+    """
+    return next((item for item in data if item is not None), None)
+
+
+def pull_dag_upstream(key, task_ids=None):
+    """
+
+    :param key:
+    :param task_ids:
+    :return:
+    """
+    context = get_current_context()
+    return select_first_non_none(
+        context[TASK_INSTANCE].xcom_pull(key=str(key),
+                                         task_ids=task_ids if task_ids else context['task'].upstream_task_ids))
+
+
+def push_dag_downstream(key, value):
+    """
+
+    :param key:
+    :param value:
+    :return:
+    """
+    context = get_current_context()
+    return context[TASK_INSTANCE].xcom_push(key=str(key), value=value)

dags/selector_daily_fetch_orchestrator.py

@@ -0,0 +1,49 @@
+from dags import DEFAULT_DAG_ARGUMENTS
+from airflow.decorators import dag, task
+from airflow.operators.python import get_current_context
+from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+from pymongo import MongoClient
+
+from ted_sws import config
+from ted_sws.data_manager.adapters.notice_repository import NoticeRepository
+from ted_sws.notice_fetcher.adapters.ted_api import TedAPIAdapter, TedRequestAPI
+from ted_sws.notice_fetcher.services.notice_fetcher import NoticeFetcher
+from ted_sws.core.model.notice import NoticeStatus
+
+@dag(default_args=DEFAULT_DAG_ARGUMENTS, tags=['selector', 'daily-fetch'])
+def selector_daily_fetch_orchestrator():
+    @task
+    def fetch_notice_from_ted():
+        context = get_current_context()
+        dag_conf = context["dag_run"].conf
+        key = 'fetch_time_filter'
+        if key in dag_conf.keys():
+            print(dag_conf[key])
+            mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+            NoticeFetcher(notice_repository=NoticeRepository(mongodb_client=mongodb_client),
+                          ted_api_adapter=TedAPIAdapter(request_api=TedRequestAPI())).fetch_notices_by_date_wild_card(
+                wildcard_date=dag_conf[key]) #"20220203*"
+        else:
+            print(f"The key={key} is not present in context")
+
+
+
+    @task
+    def trigger_document_proc_pipeline():
+        context = get_current_context()
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        notices = notice_repository.get_notice_by_status(notice_status=NoticeStatus.RAW)
+        for notice in notices:
+            TriggerDagRunOperator(
+                task_id=f'trigger_worker_dag_{notice.ted_id}',
+                trigger_dag_id="worker_single_notice_process_orchestrator",
+                conf={"notice_id": notice.ted_id,
+                      "notice_status": str(notice.status)
+                      }
+            ).execute(context=context)
+
+    fetch_notice_from_ted() >> trigger_document_proc_pipeline()
+
+
+dag = selector_daily_fetch_orchestrator()

dags/selector_repackage_process_orchestrator.py

@@ -0,0 +1,44 @@
+from dags import DEFAULT_DAG_ARGUMENTS
+from airflow.decorators import dag, task
+from airflow.operators.python import get_current_context
+from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+from pymongo import MongoClient
+from ted_sws.core.model.notice import NoticeStatus
+
+from ted_sws import config
+from ted_sws.data_manager.adapters.notice_repository import NoticeRepository
+
+RE_PACKAGE_TARGET_NOTICE_STATES = [NoticeStatus.ELIGIBLE_FOR_PACKAGING, NoticeStatus.INELIGIBLE_FOR_PUBLISHING]
+
+
+@dag(default_args=DEFAULT_DAG_ARGUMENTS, tags=['selector', 're-package'])
+def selector_re_package_process_orchestrator():
+    @task
+    def select_notices_for_re_package_and_reset_status():
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        for target_notice_state in RE_PACKAGE_TARGET_NOTICE_STATES:
+            notices = notice_repository.get_notice_by_status(notice_status=target_notice_state)
+            for notice in notices:
+                notice.update_status_to(new_status=NoticeStatus.ELIGIBLE_FOR_PACKAGING)
+                notice_repository.update(notice=notice)
+
+    @task
+    def trigger_worker_for_package_branch():
+        context = get_current_context()
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        notices = notice_repository.get_notice_by_status(notice_status=NoticeStatus.ELIGIBLE_FOR_PACKAGING)
+        for notice in notices:
+            TriggerDagRunOperator(
+                task_id=f'trigger_worker_dag_{notice.ted_id}',
+                trigger_dag_id="worker_single_notice_process_orchestrator",
+                conf={"notice_id": notice.ted_id,
+                      "notice_status": str(notice.status)
+                      }
+            ).execute(context=context)
+
+    select_notices_for_re_package_and_reset_status() >> trigger_worker_for_package_branch()
+
+
+etl_dag = selector_re_package_process_orchestrator()

dags/selector_republish_process_orchestrator.py

@@ -0,0 +1,44 @@
+from dags import DEFAULT_DAG_ARGUMENTS
+from airflow.decorators import dag, task
+from airflow.operators.python import get_current_context
+from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+from pymongo import MongoClient
+from ted_sws.core.model.notice import NoticeStatus
+
+from ted_sws import config
+from ted_sws.data_manager.adapters.notice_repository import NoticeRepository
+
+RE_PUBLISH_TARGET_NOTICE_STATES = [NoticeStatus.PUBLICLY_UNAVAILABLE, NoticeStatus.ELIGIBLE_FOR_PUBLISHING]
+
+
+@dag(default_args=DEFAULT_DAG_ARGUMENTS, tags=['selector', 're-publish'])
+def selector_re_publish_process_orchestrator():
+    @task
+    def select_notices_for_re_publish_and_reset_status():
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        for target_notice_state in RE_PUBLISH_TARGET_NOTICE_STATES:
+            notices = notice_repository.get_notice_by_status(notice_status=target_notice_state)
+            for notice in notices:
+                notice.update_status_to(new_status=NoticeStatus.ELIGIBLE_FOR_PUBLISHING)
+                notice_repository.update(notice=notice)
+
+    @task
+    def trigger_worker_for_publish_branch():
+        context = get_current_context()
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        notices = notice_repository.get_notice_by_status(notice_status=NoticeStatus.ELIGIBLE_FOR_PUBLISHING)
+        for notice in notices:
+            TriggerDagRunOperator(
+                task_id=f'trigger_worker_dag_{notice.ted_id}',
+                trigger_dag_id="worker_single_notice_process_orchestrator",
+                conf={"notice_id": notice.ted_id,
+                      "notice_status": str(notice.status)
+                      }
+            ).execute(context=context)
+
+    select_notices_for_re_publish_and_reset_status() >> trigger_worker_for_publish_branch()
+
+
+etl_dag = selector_re_publish_process_orchestrator()

dags/selector_retransform_process_orchestrator.py

@@ -0,0 +1,50 @@
+from dags import DEFAULT_DAG_ARGUMENTS
+from airflow.decorators import dag, task
+from airflow.operators.python import get_current_context
+from airflow.operators.trigger_dagrun import TriggerDagRunOperator
+from pymongo import MongoClient
+from ted_sws.core.model.notice import NoticeStatus
+
+from ted_sws import config
+from ted_sws.data_manager.adapters.notice_repository import NoticeRepository
+
+RE_TRANSFORM_TARGET_NOTICE_STATES = [NoticeStatus.ELIGIBLE_FOR_TRANSFORMATION, NoticeStatus.NORMALISED_METADATA,
+                                     NoticeStatus.ELIGIBLE_FOR_TRANSFORMATION,
+                                     NoticeStatus.PREPROCESSED_FOR_TRANSFORMATION,
+                                     NoticeStatus.INELIGIBLE_FOR_TRANSFORMATION, NoticeStatus.TRANSFORMED,
+                                     NoticeStatus.DISTILLED,
+                                     NoticeStatus.VALIDATED, NoticeStatus.INELIGIBLE_FOR_PACKAGING
+                                     ]
+
+
+@dag(default_args=DEFAULT_DAG_ARGUMENTS, tags=['selector', 're-transform'])
+def selector_re_transform_process_orchestrator():
+    @task
+    def select_notices_for_re_transform_and_reset_status():
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        for target_notice_state in RE_TRANSFORM_TARGET_NOTICE_STATES:
+            notices = notice_repository.get_notice_by_status(notice_status=target_notice_state)
+            for notice in notices:
+                notice.update_status_to(new_status=NoticeStatus.ELIGIBLE_FOR_TRANSFORMATION)
+                notice_repository.update(notice=notice)
+
+    @task
+    def trigger_worker_for_transform_branch():
+        context = get_current_context()
+        mongodb_client = MongoClient(config.MONGO_DB_AUTH_URL)
+        notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+        notices = notice_repository.get_notice_by_status(notice_status=NoticeStatus.ELIGIBLE_FOR_TRANSFORMATION)
+        for notice in notices:
+            TriggerDagRunOperator(
+                task_id=f'trigger_worker_dag_{notice.ted_id}',
+                trigger_dag_id="worker_single_notice_process_orchestrator",
+                conf={"notice_id": notice.ted_id,
+                      "notice_status": str(notice.status)
+                      }
+            ).execute(context=context)
+
+    select_notices_for_re_transform_and_reset_status() >> trigger_worker_for_transform_branch()
+
+
+etl_dag = selector_re_transform_process_orchestrator()