Merge pull request #476 from OP-TED/feature/TED-1365

Feature/ted 1365

Author: CaptainOfHacks

dags/fetch_notices_by_date.py

@@ -13,6 +13,7 @@
 from ted_sws.event_manager.adapters.event_log_decorator import event_log
 from ted_sws.event_manager.model.event_message import TechnicalEventMessage, EventMessageMetadata, \
     EventMessageProcessType
+from ted_sws.event_manager.services.log import log_error
 
 DAG_NAME = "fetch_notices_by_date"
 BATCH_SIZE = 2000
@@ -40,8 +41,9 @@ 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))
         if not notice_ids:
-            raise Exception("No notices has been fetched!")
-        push_dag_downstream(key=NOTICE_IDS_KEY, value=notice_ids)
+            log_error("No notices has been fetched!")
+        else:
+            push_dag_downstream(key=NOTICE_IDS_KEY, value=notice_ids)
 
     trigger_complete_workflow = TriggerNoticeBatchPipelineOperator(task_id=TRIGGER_COMPLETE_WORKFLOW_TASK_ID,
                                                                    execute_only_one_step=False
@@ -67,9 +69,13 @@ def validate_fetched_notices():
         validate_and_update_daily_supra_notice(ted_publication_date=publication_date, mongodb_client=mongodb_client)
 
     def _branch_selector():
+        notice_ids = pull_dag_upstream(key=NOTICE_IDS_KEY)
+        if not notice_ids:
+            return [FINISH_FETCH_BY_DATE_TASK_ID]
+
         trigger_complete_workflow = get_dag_param(key=TRIGGER_COMPLETE_WORKFLOW_DAG_KEY,
                                                   default_value=True)
-        push_dag_downstream(key=NOTICE_IDS_KEY, value=pull_dag_upstream(key=NOTICE_IDS_KEY))
+        push_dag_downstream(key=NOTICE_IDS_KEY, value=notice_ids)
         if trigger_complete_workflow:
             return [TRIGGER_COMPLETE_WORKFLOW_TASK_ID]
         return [TRIGGER_PARTIAL_WORKFLOW_TASK_ID]
@@ -91,5 +97,7 @@ def _branch_selector():
     fetch_by_date_notice_from_ted() >> branch_task >> [trigger_normalisation_workflow,
                                                        trigger_complete_workflow] >> validate_fetched_notices_step >> finish_step
 
+    branch_task >> finish_step
+
 
 dag = fetch_notices_by_date()

requirements.txt

@@ -1,6 +1,6 @@
 # Project dependecies
 pydantic~=1.9.0
-requests~=2.28.1
+requests~=2.28.2
 deepdiff~=5.7.0
 jinja2~=3.1.2
 python-dotenv~=0.19.2
@@ -25,4 +25,4 @@ ordered-set~=4.0.2
 json2html~=1.3.0
 minio~=7.1.1
 certifi~=2022.12.7
-shortuuid~=1.0.11
+shortuuid~=1.0.11