rename DAGs

Author: CaptainOfHacks

dags/daily_materialized_view_update.py dags/daily_materialized_views_update.pydags/daily_materialized_view_update.py renamed to dags/daily_materialized_views_update.py

@@ -8,14 +8,14 @@
 from ted_sws.data_manager.services.create_notice_collection_materialised_view import \
     create_notice_collection_materialised_view, create_notice_kpi_collection
 
-DAG_NAME = "daily_materialized_view_update"
+DAG_NAME = "daily_materialized_views_update"
 
 
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
      catchup=False,
      schedule_interval="0 6 * * *",
      tags=['mongodb', 'daily-views-update'])
-def daily_materialized_view_update():
+def daily_materialized_views_update():
     @task
     def create_materialised_view():
         mongo_client = MongoClient(config.MONGO_DB_AUTH_URL)
@@ -34,4 +34,4 @@ def aggregate_batch_logs():
     create_materialised_view() >> create_kpi_collection_for_notices() >> aggregate_batch_logs()
 
 
-dag = daily_materialized_view_update()
+dag = daily_materialized_views_update()

dags/notice_fetch_by_date_workflow.py dags/fetch_notices_by_date.pydags/notice_fetch_by_date_workflow.py renamed to dags/fetch_notices_by_date.py

@@ -14,7 +14,7 @@
 from ted_sws.event_manager.model.event_message import TechnicalEventMessage, EventMessageMetadata, \
     EventMessageProcessType
 
-DAG_NAME = "notice_fetch_by_date_workflow"
+DAG_NAME = "fetch_notices_by_date"
 BATCH_SIZE = 2000
 WILD_CARD_DAG_KEY = "wild_card"
 TRIGGER_COMPLETE_WORKFLOW_DAG_KEY = "trigger_complete_workflow"
@@ -29,7 +29,7 @@
      catchup=False,
      timetable=CronTriggerTimetable('0 1 * * *', timezone='UTC'),
      tags=['selector', 'daily-fetch'])
-def notice_fetch_by_date_workflow():
+def fetch_notices_by_date():
     @task
     @event_log(TechnicalEventMessage(
         message="fetch_notice_from_ted",
@@ -93,4 +93,4 @@ def _branch_selector():
                                                        trigger_complete_workflow] >> validate_fetched_notices_step >> finish_step
 
 
-dag = notice_fetch_by_date_workflow()
+dag = fetch_notices_by_date()

…ice_fetch_for_date_range_orchestrator.py dags/fetch_notices_by_date_range.pydags/notice_fetch_for_date_range_orchestrator.py renamed to dags/fetch_notices_by_date_range.py dags/notice_fetch_for_date_range_orchestrator.py renamed to dags/fetch_notices_by_date_range.py

@@ -8,12 +8,12 @@
 
 from dags import DEFAULT_DAG_ARGUMENTS
 from dags.dags_utils import get_dag_param
-from dags.notice_fetch_by_date_workflow import WILD_CARD_DAG_KEY, TRIGGER_COMPLETE_WORKFLOW_DAG_KEY
+from dags.fetch_notices_by_date import WILD_CARD_DAG_KEY, TRIGGER_COMPLETE_WORKFLOW_DAG_KEY
 from ted_sws.event_manager.adapters.event_log_decorator import event_log
 from ted_sws.event_manager.model.event_message import TechnicalEventMessage, EventMessageMetadata, \
     EventMessageProcessType
 
-DAG_NAME = "notice_fetch_for_date_range_orchestrator"
+DAG_NAME = "fetch_notices_by_date_range"
 
 START_DATE_KEY = "start_date"
 END_DATE_KEY = "end_date"
@@ -33,7 +33,7 @@ def generate_wildcards_foreach_day_in_range(start_date: str, end_date: str) -> l
 
 
 @dag(default_args=DEFAULT_DAG_ARGUMENTS, schedule_interval=None, tags=['master'])
-def notice_fetch_for_date_range_orchestrator():
+def fetch_notices_by_date_range():
     @task
     @event_log(TechnicalEventMessage(
         message="trigger_fetch_notices_workers_for_date_range",
@@ -59,4 +59,4 @@ def trigger_notice_by_date_for_each_date_in_range():
     trigger_notice_by_date_for_each_date_in_range()
 
 
-dag = notice_fetch_for_date_range_orchestrator()
+dag = fetch_notices_by_date_range()

dags/load_mapping_suite_in_mongodb.py dags/load_mapping_suite_in_database.pydags/load_mapping_suite_in_mongodb.py renamed to dags/load_mapping_suite_in_database.py

@@ -31,7 +31,7 @@
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
      schedule_interval=None,
      tags=['fetch', 'mapping-suite', 'github'])
-def load_mapping_suite_in_mongodb():
+def load_mapping_suite_in_database():
     @task
     @event_log(is_loggable=False)
     def fetch_mapping_suite_package_from_github_into_mongodb(**context_args):
@@ -86,4 +86,4 @@ def _branch_selector():
     branch_task >> [trigger_document_proc_pipeline, finish_step]
 
 
-dag = load_mapping_suite_in_mongodb()
+dag = load_mapping_suite_in_database()

dags/notice_process_workflow.py dags/notice_processing_pipeline.pydags/notice_process_workflow.py renamed to dags/notice_processing_pipeline.py

@@ -24,7 +24,7 @@
 SELECTOR_BRANCH_BEFORE_VALIDATION_TASK_ID = "switch_to_validation"
 SELECTOR_BRANCH_BEFORE_PACKAGE_TASK_ID = "switch_to_package"
 SELECTOR_BRANCH_BEFORE_PUBLISH_TASK_ID = "switch_to_publish"
-DAG_NAME = "notice_process_workflow"
+DAG_NAME = "notice_processing_pipeline"
 
 BRANCH_SELECTOR_MAP = {NOTICE_NORMALISATION_PIPELINE_TASK_ID: NOTICE_NORMALISATION_PIPELINE_TASK_ID,
                        NOTICE_TRANSFORMATION_PIPELINE_TASK_ID: SELECTOR_BRANCH_BEFORE_TRANSFORMATION_TASK_ID,
@@ -49,7 +49,7 @@ def branch_selector(result_branch: str, xcom_forward_keys: List[str] = [NOTICE_I
      max_active_runs=256,
      max_active_tasks=256,
      tags=['worker', 'pipeline'])
-def notice_process_workflow():
+def notice_processing_pipeline():
     """
 
     """
@@ -149,4 +149,4 @@ def _stop_processing():
     notice_package_step >> selector_branch_before_publish >> notice_publish_step
 
 
-dag = notice_process_workflow()
+dag = notice_processing_pipeline()

…ctor_raw_notices_process_orchestrator.py …ess_unnormalised_notices_from_backlog.pydags/selector_raw_notices_process_orchestrator.py renamed to dags/reprocess_unnormalised_notices_from_backlog.py dags/selector_raw_notices_process_orchestrator.py renamed to dags/reprocess_unnormalised_notices_from_backlog.py

@@ -8,7 +8,7 @@
 from ted_sws.event_manager.model.event_message import TechnicalEventMessage, EventMessageMetadata, \
     EventMessageProcessType
 
-DAG_NAME = "selector_raw_notices_process_orchestrator"
+DAG_NAME = "reprocess_unnormalised_notices_from_backlog"
 
 TRIGGER_NOTICE_PROCESS_WORKFLOW_TASK_ID = "trigger_notice_process_workflow"
 FORM_NUMBER_DAG_PARAM = "form_number"
@@ -20,7 +20,7 @@
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
      schedule_interval=None,
      tags=['selector', 'raw-notices'])
-def selector_raw_notices_process_orchestrator():
+def reprocess_unnormalised_notices_from_backlog():
     @task
     @event_log(TechnicalEventMessage(
         message="select_all_raw_notices",
@@ -41,4 +41,4 @@ def select_all_raw_notices():
     select_all_raw_notices() >> trigger_notice_process_workflow
 
 
-dag = selector_raw_notices_process_orchestrator()
+dag = reprocess_unnormalised_notices_from_backlog()

…lector_repackage_process_orchestrator.py …ocess_unpackaged_notices_from_backlog.pydags/selector_repackage_process_orchestrator.py renamed to dags/reprocess_unpackaged_notices_from_backlog.py dags/selector_repackage_process_orchestrator.py renamed to dags/reprocess_unpackaged_notices_from_backlog.py

@@ -2,15 +2,15 @@
 
 from dags import DEFAULT_DAG_ARGUMENTS
 from dags.dags_utils import push_dag_downstream, get_dag_param
-from dags.notice_process_workflow import NOTICE_PACKAGE_PIPELINE_TASK_ID
+from dags.notice_processing_pipeline import NOTICE_PACKAGE_PIPELINE_TASK_ID
 from dags.operators.DagBatchPipelineOperator import NOTICE_IDS_KEY, TriggerNoticeBatchPipelineOperator
 from dags.pipelines.notice_selectors_pipelines import notice_ids_selector_by_status
 from ted_sws.core.model.notice import NoticeStatus
 from ted_sws.event_manager.adapters.event_log_decorator import event_log
 from ted_sws.event_manager.model.event_message import TechnicalEventMessage, EventMessageMetadata, \
     EventMessageProcessType
 
-DAG_NAME = "selector_re_package_process_orchestrator"
+DAG_NAME = "reprocess_unpackaged_notices_from_backlog"
 
 RE_PACKAGE_TARGET_NOTICE_STATES = [NoticeStatus.VALIDATED, NoticeStatus.INELIGIBLE_FOR_PACKAGING,
                                    NoticeStatus.ELIGIBLE_FOR_PACKAGING,
@@ -25,7 +25,7 @@
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
      schedule_interval=None,
      tags=['selector', 're-package'])
-def selector_re_package_process_orchestrator():
+def reprocess_unpackaged_notices_from_backlog():
     @task
     @event_log(TechnicalEventMessage(
         message="select_notices_for_re_package",
@@ -50,4 +50,4 @@ def select_notices_for_re_package():
     select_notices_for_re_package() >> trigger_notice_process_workflow
 
 
-dag = selector_re_package_process_orchestrator()
+dag = reprocess_unpackaged_notices_from_backlog()

…lector_republish_process_orchestrator.py …cess_unpublished_notices_from_backlog.pydags/selector_republish_process_orchestrator.py renamed to dags/reprocess_unpublished_notices_from_backlog.py dags/selector_republish_process_orchestrator.py renamed to dags/reprocess_unpublished_notices_from_backlog.py

@@ -2,7 +2,7 @@
 
 from dags import DEFAULT_DAG_ARGUMENTS
 from dags.dags_utils import push_dag_downstream, get_dag_param
-from dags.notice_process_workflow import NOTICE_PUBLISH_PIPELINE_TASK_ID
+from dags.notice_processing_pipeline import NOTICE_PUBLISH_PIPELINE_TASK_ID
 from dags.operators.DagBatchPipelineOperator import NOTICE_IDS_KEY, TriggerNoticeBatchPipelineOperator, \
     EXECUTE_ONLY_ONE_STEP_KEY
 from dags.pipelines.notice_selectors_pipelines import notice_ids_selector_by_status
@@ -11,7 +11,7 @@
 from ted_sws.event_manager.model.event_message import TechnicalEventMessage, EventMessageMetadata, \
     EventMessageProcessType
 
-DAG_NAME = "selector_re_publish_process_orchestrator"
+DAG_NAME = "reprocess_unpublished_notices_from_backlog"
 
 RE_PUBLISH_TARGET_NOTICE_STATES = [NoticeStatus.ELIGIBLE_FOR_PUBLISHING, NoticeStatus.INELIGIBLE_FOR_PUBLISHING,
                                    NoticeStatus.PACKAGED
@@ -26,7 +26,7 @@
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
      schedule_interval=None,
      tags=['selector', 're-publish'])
-def selector_re_publish_process_orchestrator():
+def reprocess_unpublished_notices_from_backlog():
     @task
     @event_log(TechnicalEventMessage(
         message="select_notices_for_re_publish",
@@ -51,4 +51,4 @@ def select_notices_for_re_publish():
     select_notices_for_re_publish() >> trigger_notice_process_workflow
 
 
-etl_dag = selector_re_publish_process_orchestrator()
+etl_dag = reprocess_unpublished_notices_from_backlog()

…ctor_retransform_process_orchestrator.py …ss_untransformed_notices_from_backlog.pydags/selector_retransform_process_orchestrator.py renamed to dags/reprocess_untransformed_notices_from_backlog.py dags/selector_retransform_process_orchestrator.py renamed to dags/reprocess_untransformed_notices_from_backlog.py

@@ -2,7 +2,7 @@
 
 from dags import DEFAULT_DAG_ARGUMENTS
 from dags.dags_utils import push_dag_downstream, get_dag_param
-from dags.notice_process_workflow import NOTICE_TRANSFORMATION_PIPELINE_TASK_ID
+from dags.notice_processing_pipeline import NOTICE_TRANSFORMATION_PIPELINE_TASK_ID
 from dags.operators.DagBatchPipelineOperator import NOTICE_IDS_KEY, TriggerNoticeBatchPipelineOperator, \
     EXECUTE_ONLY_ONE_STEP_KEY
 from dags.pipelines.notice_selectors_pipelines import notice_ids_selector_by_status
@@ -11,7 +11,7 @@
 from ted_sws.event_manager.model.event_message import TechnicalEventMessage, EventMessageMetadata, \
     EventMessageProcessType
 
-DAG_NAME = "selector_re_transform_process_orchestrator"
+DAG_NAME = "reprocess_untransformed_notices_from_backlog"
 
 RE_TRANSFORM_TARGET_NOTICE_STATES = [NoticeStatus.NORMALISED_METADATA, NoticeStatus.INELIGIBLE_FOR_TRANSFORMATION,
                                      NoticeStatus.ELIGIBLE_FOR_TRANSFORMATION, NoticeStatus.PREPROCESSED_FOR_TRANSFORMATION,
@@ -27,7 +27,7 @@
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
      schedule_interval=None,
      tags=['selector', 're-transform'])
-def selector_re_transform_process_orchestrator():
+def reprocess_untransformed_notices_from_backlog():
     @task
     @event_log(TechnicalEventMessage(
         message="select_notices_for_re_transform",
@@ -52,4 +52,4 @@ def select_notices_for_re_transform():
     select_notices_for_re_transform() >> trigger_notice_process_workflow
 
 
-dag = selector_re_transform_process_orchestrator()
+dag = reprocess_untransformed_notices_from_backlog()

…ector_revalidate_process_orchestrator.py …cess_unvalidated_notices_from_backlog.pydags/selector_revalidate_process_orchestrator.py renamed to dags/reprocess_unvalidated_notices_from_backlog.py dags/selector_revalidate_process_orchestrator.py renamed to dags/reprocess_unvalidated_notices_from_backlog.py

@@ -2,7 +2,7 @@
 
 from dags import DEFAULT_DAG_ARGUMENTS
 from dags.dags_utils import push_dag_downstream, get_dag_param
-from dags.notice_process_workflow import NOTICE_TRANSFORMATION_PIPELINE_TASK_ID
+from dags.notice_processing_pipeline import NOTICE_TRANSFORMATION_PIPELINE_TASK_ID
 from dags.operators.DagBatchPipelineOperator import NOTICE_IDS_KEY, TriggerNoticeBatchPipelineOperator, \
     EXECUTE_ONLY_ONE_STEP_KEY
 from dags.pipelines.notice_selectors_pipelines import notice_ids_selector_by_status
@@ -11,7 +11,7 @@
 from ted_sws.event_manager.model.event_message import TechnicalEventMessage, EventMessageMetadata, \
     EventMessageProcessType
 
-DAG_NAME = "selector_re_validate_process_orchestrator"
+DAG_NAME = "reprocess_unvalidated_notices_from_backlog"
 
 RE_VALIDATE_TARGET_NOTICE_STATES = [NoticeStatus.DISTILLED]
 TRIGGER_NOTICE_PROCESS_WORKFLOW_TASK_ID = "trigger_notice_process_workflow"
@@ -24,7 +24,7 @@
 @dag(default_args=DEFAULT_DAG_ARGUMENTS,
      schedule_interval=None,
      tags=['selector', 're-validate'])
-def selector_re_validate_process_orchestrator():
+def reprocess_unvalidated_notices_from_backlog():
     @task
     @event_log(TechnicalEventMessage(
         message="select_notices_for_re_validate",
@@ -49,4 +49,4 @@ def select_notices_for_re_validate():
     select_notices_for_re_validate() >> trigger_notice_process_workflow
 
 
-dag = selector_re_validate_process_orchestrator()
+dag = reprocess_unvalidated_notices_from_backlog()