BranchDateTimeOperator¶
Use the BranchDateTimeOperator to branch into one of two execution paths depending on whether the date and/or time of execution falls into the range given by two target arguments.
empty_task_11 = EmptyOperator(task_id='date_in_range', dag=dag1)
empty_task_21 = EmptyOperator(task_id='date_outside_range', dag=dag1)
cond1 = BranchDateTimeOperator(
task_id='datetime_branch',
follow_task_ids_if_true=['date_in_range'],
follow_task_ids_if_false=['date_outside_range'],
target_upper=pendulum.datetime(2020, 10, 10, 15, 0, 0),
target_lower=pendulum.datetime(2020, 10, 10, 14, 0, 0),
dag=dag1,
)
# Run empty_task_1 if cond1 executes between 2020-10-10 14:00:00 and 2020-10-10 15:00:00
cond1 >> [empty_task_11, empty_task_21]
The target parameters, target_upper and target_lower, can receive a datetime.datetime, a datetime.time, or None. When a datetime.time object is used, it will be combined with the current date in order to allow comparisons with it. In the event that target_upper is set to a datetime.time that occurs before the given target_lower, a day will be added to target_upper. This is done to allow for time periods that span over two dates.
empty_task_12 = EmptyOperator(task_id='date_in_range', dag=dag2)
empty_task_22 = EmptyOperator(task_id='date_outside_range', dag=dag2)
cond2 = BranchDateTimeOperator(
task_id='datetime_branch',
follow_task_ids_if_true=['date_in_range'],
follow_task_ids_if_false=['date_outside_range'],
target_upper=pendulum.time(0, 0, 0),
target_lower=pendulum.time(15, 0, 0),
dag=dag2,
)
# Since target_lower happens after target_upper, target_upper will be moved to the following day
# Run empty_task_1 if cond2 executes between 15:00:00, and 00:00:00 of the following day
cond2 >> [empty_task_12, empty_task_22]
If a target parameter is set to None, the operator will perform a unilateral comparison using only the non-None target. Setting both target_upper and target_lower to None will raise an exception.