Skip to content

Commit ed05bcb

Browse files
committed
implement scheduling for daily materialised view update + make default dag timezone global
1 parent b954983 commit ed05bcb

3 files changed

Lines changed: 9 additions & 6 deletions

File tree

dags/daily_materialized_views_update.py

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,9 @@
11
from airflow.decorators import dag, task
2+
from airflow.timetables.trigger import CronTriggerTimetable
23
from pymongo import MongoClient
34

45
from dags import DEFAULT_DAG_ARGUMENTS
5-
from ted_sws import config
6+
from ted_sws import config, DAG_DEFAULT_TIMEZONE
67
from ted_sws.data_manager.services.create_batch_collection_materialised_view import \
78
create_batch_collection_materialised_view
89
from ted_sws.data_manager.services.create_notice_collection_materialised_view import \
@@ -12,8 +13,11 @@
1213

1314

1415
@dag(default_args=DEFAULT_DAG_ARGUMENTS,
16+
dag_id=DAILY_MATERIALISED_VIEWS_DAG_NAME,
1517
catchup=False,
16-
schedule_interval="0 6 * * *",
18+
timetable=CronTriggerTimetable(
19+
cron=config.SCHEDULE_DAG_MATERIALIZED_VIEW_UPDATE,
20+
timezone=DAG_DEFAULT_TIMEZONE),
1721
tags=['mongodb', 'daily-views-update'])
1822
def daily_materialized_views_update():
1923
@task

dags/fetch_notices_by_date.py

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
from dags.dags_utils import get_dag_param, push_dag_downstream, pull_dag_upstream
1111
from dags.operators.DagBatchPipelineOperator import NOTICE_IDS_KEY, TriggerNoticeBatchPipelineOperator
1212
from dags.pipelines.notice_fetcher_pipelines import notice_fetcher_by_date_pipeline
13-
from ted_sws import config
13+
from ted_sws import config, DAG_DEFAULT_TIMEZONE
1414
from ted_sws.event_manager.adapters.event_log_decorator import event_log
1515
from ted_sws.event_manager.model.event_message import TechnicalEventMessage, EventMessageMetadata, \
1616
EventMessageProcessType
@@ -26,15 +26,13 @@
2626
FINISH_FETCH_BY_DATE_TASK_ID = "finish_fetch_by_date"
2727
VALIDATE_FETCHED_NOTICES_TASK_ID = "validate_fetched_notices"
2828

29-
DAG_FETCH_DEFAULT_TIMEZONE = "UTC"
30-
3129

3230
@dag(default_args=DEFAULT_DAG_ARGUMENTS,
3331
dag_id=FETCHER_DAG_NAME,
3432
catchup=False,
3533
timetable=CronTriggerTimetable(
3634
cron=config.SCHEDULE_DAG_FETCH,
37-
timezone=DAG_FETCH_DEFAULT_TIMEZONE),
35+
timezone=DAG_DEFAULT_TIMEZONE),
3836
tags=['selector', 'daily-fetch'])
3937
def fetch_notices_by_date():
4038
@task

ted_sws/__init__.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737

3838
DAG_FETCH_DEFAULT_TIMETABLE = "0 1 * * *"
3939
DAG_MATERIALIZED_VIEW_UPDATE_DEFAULT_TIMETABLE = "0 6 * * *"
40+
DAG_DEFAULT_TIMEZONE = "UTC"
4041

4142
class MongoDBConfig:
4243

0 commit comments

Comments
 (0)