Repository: apache/airflow
Pull request: Add new datetime branch operator
closes: #11929 

This PR includes a new datetime branching operator: the current date and time, as given by `datetime.datetime.now` is compared against target `datetime` attributes, like `year` or `hour`, to decide which task id branch to take.

No tests were written yet as this is intended as a POC to allow us to review the implementation first. 

**^ Add meaningful description above**

Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).


Problem statement:
add DateTimeBranchOperator

**Use case / motivation**

Airflow have `BranchSQLOperator` and `BranchPythonOperator`.
Airflow also have `DateTimeSensor`.

I find it very useful to have `DateTimeBranchOperator` that will allow to branch workflow based on specific DateTime
Sometimes in DAGs you want to do another set of tasks according to specific dates or time.

Hints from issue discussion:
Hello 👋 

I've been using Airflow for a while now and would love to get involved with the project and contribute some code. If nobody's working on this I'd like to open a PR. Just wanted to clarify a few questions:

* Should `DateTimeBranchOperator` support comparisons with partial dates? For example: branching based on a comparison between the current date and only a year, instead of a full date. I believe we should, although this may lead to a long argument list as we would need one for each date attribute.
* Should the operator inherit from `BaseBranchOperator`? Seems to me that would be the best option, but none of the examples mentioned inherit from it (`BranchSQLOperator` and `BranchPythonOperator`).
@tomasfarias Hi. It is best if you prepare a small POC that will not contain tests and will not even fully work but will allow us to familiarize ourselves with your concept. Then we can iterate over this code together. What do you think about it?
Sounds perfect to me! Thanks for the guidance @mik-laj, I'll open a PR shortly.

Commit message: Update example dag description

Patch to review:
diff --git a/airflow/example_dags/example_datetime_branch_operator.py b/airflow/example_dags/example_datetime_branch_operator.py
new file mode 100644
index 0000000000000..58b1a0a5fc579
--- /dev/null
+++ b/airflow/example_dags/example_datetime_branch_operator.py
@@ -0,0 +1,83 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Example DAG demonstrating the usage of DateTimeBranchOperator with datetime as well as time objects as
+targets.
+"""
+import datetime
+
+from airflow import DAG
+from airflow.operators.datetime_branch import DateTimeBranchOperator
+from airflow.operators.dummy_operator import DummyOperator
+from airflow.utils.dates import days_ago
+
+args = {
+    "owner": "airflow",
+}
+
+dag = DAG(
+    dag_id="example_datetime_branch_operator",
+    start_date=days_ago(2),
+    default_args=args,
+    tags=["example"],
+    schedule_interval="@daily",
+)
+
+# [START howto_operator_datetime_branch]
+dummy_task_1 = DummyOperator(task_id='date_in_range', dag=dag)
+dummy_task_2 = DummyOperator(task_id='date_outside_range', dag=dag)
+
+cond1 = DateTimeBranchOperator(
+    task_id='datetime_branch',
+    follow_task_ids_if_true=['date_in_range'],
+    follow_task_ids_if_false=['date_outside_range'],
+    target_upper=datetime.datetime(2020, 10, 10, 15, 0, 0),
+    target_lower=datetime.datetime(2020, 10, 10, 14, 0, 0),
+    dag=dag,
+)
+
+# Run dummy_task_1 if cond1 executes between 2020-10-10 14:00:00 and 2020-10-10 15:00:00
+cond1 >> [dummy_task_1, dummy_task_2]
+# [END howto_operator_datetime_branch]
+
+
+dag = DAG(
+    dag_id="example_datetime_branch_operator_2",
+    start_date=days_ago(2),
+    default_args=args,
+    tags=["example"],
+    schedule_interval="@daily",
+)
+# [START howto_operator_datetime_branch_next_day]
+dummy_task_1 = DummyOperator(task_id='date_in_range', dag=dag)
+dummy_task_2 = DummyOperator(task_id='date_outside_range', dag=dag)
+
+cond2 = DateTimeBranchOperator(
+    task_id='datetime_branch',
+    follow_task_ids_if_true=['date_in_range'],
+    follow_task_ids_if_false=['date_outside_range'],
+    target_upper=datetime.time(0, 0, 0),
+    target_lower=datetime.time(15, 0, 0),
+    dag=dag,
+)
+
+# Since target_lower happens after target_upper, target_upper will be moved to the following day
+# Run dummy_task_1 if cond2 executes between 15:00:00, and 00:00:00 of the following day
+cond2 >> [dummy_task_1, dummy_task_2]
+# [END howto_operator_datetime_branch_next_day]
diff --git a/airflow/operators/datetime_branch.py b/airflow/operators/datetime_branch.py
new file mode 100644
index 0000000000000..9a2fd962d0837
--- /dev/null
+++ b/airflow/operators/datetime_branch.py
@@ -0,0 +1,101 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import datetime
+from typing import Dict, Iterable, Union
+
+from airflow.exceptions import AirflowException
+from airflow.operators.branch_operator import BaseBranchOperator
+from airflow.utils import timezone
+from airflow.utils.decorators import apply_defaults
+
+
+class DateTimeBranchOperator(BaseBranchOperator):
+    """
+    Branches into one of two lists of tasks depending on the current datetime.
+
+    True branch will be returned when `datetime.datetime.now()` falls below
+    `target_upper` and above `target_lower`.
+
+    :param follow_task_ids_if_true: task id or task ids to follow if
+        `datetime.datetime.now()` falls above target_lower and below `target_upper`.
+    :type follow_task_ids_if_true: str or list[str]
+    :param follow_task_ids_if_false: task id or task ids to follow if
+        `datetime.datetime.now()` falls below target_lower or above `target_upper`.
+    :type follow_task_ids_if_false: str or list[str]
+    :param target_lower: target lower bound.
+    :type target_lower: Optional[datetime.datetime]
+    :param target_upper: target upper bound.
+    :type target_upper: Optional[datetime.datetime]
+    """
+
+    @apply_defaults
+    def __init__(
+        self,
+        *,
+        follow_task_ids_if_true: Union[str, Iterable[str]],
+        follow_task_ids_if_false: Union[str, Iterable[str]],
+        target_lower: Union[datetime.datetime, datetime.time, None],
+        target_upper: Union[datetime.datetime, datetime.time, None],
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        if target_lower is None and target_upper is None:
+            raise AirflowException(
+                "Both target_upper and target_lower are None. At least one "
+                "must be defined to be compared to the current datetime"
+            )
+
+        self.target_lower = target_lower
+        self.target_upper = target_upper
+        self.follow_task_ids_if_true = follow_task_ids_if_true
+        self.follow_task_ids_if_false = follow_task_ids_if_false
+
+    def choose_branch(self, context: Dict) -> Union[str, Iterable[str]]:
+        now = timezone.make_naive(timezone.utcnow(), self.dag.timezone)
+
+        lower, upper = target_times_as_dates(now, self.target_lower, self.target_upper)
+
+        if upper is not None and upper < now:
+            return self.follow_task_ids_if_false
+
+        if lower is not None and lower > now:
+            return self.follow_task_ids_if_false
+
+        return self.follow_task_ids_if_true
+
+
+def target_times_as_dates(
+    base_date: datetime.datetime,
+    lower: Union[datetime.datetime, datetime.time, None],
+    upper: Union[datetime.datetime, datetime.time, None],
+):
+    """Ensures upper and lower time targets are datetimes by combining them with base_date"""
+    if isinstance(lower, datetime.datetime) and isinstance(upper, datetime.datetime):
+        return lower, upper
+
+    if lower is not None and isinstance(lower, datetime.time):
+        lower = datetime.datetime.combine(base_date, lower)
+    if upper is not None and isinstance(upper, datetime.time):
+        upper = datetime.datetime.combine(base_date, upper)
+
+    if any(date is None for date in (lower, upper)):
+        return lower, upper
+
+    if upper < lower:
+        upper += datetime.timedelta(days=1)
+    return lower, upper
diff --git a/docs/apache-airflow/howto/operator/datetime_branch.rst b/docs/apache-airflow/howto/operator/datetime_branch.rst
new file mode 100644
index 0000000000000..b798468d23623
--- /dev/null
+++ b/docs/apache-airflow/howto/operator/datetime_branch.rst
@@ -0,0 +1,39 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+
+
+.. _howto/operator:DatetimeBranch:
+
+DatetimeBranchOperator
+======================
+
+Use the :class:`~airflow.operators.datetime_branch.DatetimeBranchOperator` 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.
+
+.. exampleinclude:: /../../airflow/example_dags/example_datetime_branch_operator.py
+    :language: python
+    :start-after: [START howto_operator_datetime_branch]
+    :end-before: [END howto_operator_datetime_branch]
+
+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.
+
+.. exampleinclude:: /../../airflow/example_dags/example_datetime_branch_operator.py
+    :language: python
+    :start-after: [START howto_operator_datetime_branch_next_day]
+    :end-before: [END howto_operator_datetime_branch_next_day]
+
+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.
diff --git a/docs/apache-airflow/howto/operator/index.rst b/docs/apache-airflow/howto/operator/index.rst
index 89ad5f3e5b884..4f615ccb96a21 100644
--- a/docs/apache-airflow/howto/operator/index.rst
+++ b/docs/apache-airflow/howto/operator/index.rst
@@ -32,5 +32,6 @@ determine what actually executes when your DAG runs.
     :maxdepth: 2
 
     bash
+    datetime_branch
     python
     external_task_sensor
diff --git a/tests/operators/test_datetime_branch.py b/tests/operators/test_datetime_branch.py
new file mode 100644
index 0000000000000..69e8081e7ae42
--- /dev/null
+++ b/tests/operators/test_datetime_branch.py
@@ -0,0 +1,226 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import datetime
+import unittest
+
+import freezegun
+
+from airflow.exceptions import AirflowException
+from airflow.models import DAG, DagRun, TaskInstance as TI
+from airflow.operators.datetime_branch import DateTimeBranchOperator
+from airflow.operators.dummy_operator import DummyOperator
+from airflow.utils import timezone
+from airflow.utils.session import create_session
+from airflow.utils.state import State
+
+DEFAULT_DATE = timezone.datetime(2016, 1, 1)
+INTERVAL = datetime.timedelta(hours=12)
+
+
+class TestDateTimeBranchOperator(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+
+        with create_session() as session:
+            session.query(DagRun).delete()
+            session.query(TI).delete()
+
+        cls.targets = [
+            (datetime.datetime(2020, 7, 7, 10, 0, 0), datetime.datetime(2020, 7, 7, 11, 0, 0)),
+            (datetime.time(10, 0, 0), datetime.time(11, 0, 0)),
+            (datetime.datetime(2020, 7, 7, 10, 0, 0), datetime.time(11, 0, 0)),
+            (datetime.time(10, 0, 0), datetime.datetime(2020, 7, 7, 11, 0, 0)),
+            (datetime.time(11, 0, 0), datetime.time(10, 0, 0)),
+        ]
+
+    def setUp(self):
+        self.dag = DAG(
+            'datetime_branch_operator_test',
+            default_args={'owner': 'airflow', 'start_date': DEFAULT_DATE},
+            schedule_interval=INTERVAL,
+        )
+
+        self.branch_1 = DummyOperator(task_id='branch_1', dag=self.dag)
+        self.branch_2 = DummyOperator(task_id='branch_2', dag=self.dag)
+
+        self.branch_op = DateTimeBranchOperator(
+            task_id='datetime_branch',
+            follow_task_ids_if_true='branch_1',
+            follow_task_ids_if_false='branch_2',
+            target_upper=datetime.datetime(2020, 7, 7, 11, 0, 0),
+            target_lower=datetime.datetime(2020, 7, 7, 10, 0, 0),
+            dag=self.dag,
+        )
+
+        self.branch_1.set_upstream(self.branch_op)
+        self.branch_2.set_upstream(self.branch_op)
+        self.dag.clear()
+
+        self.dr = self.dag.create_dagrun(
+            run_id='manual__', start_date=DEFAULT_DATE, execution_date=DEFAULT_DATE, state=State.RUNNING
+        )
+
+    def tearDown(self):
+        super().tearDown()
+
+        with create_session() as session:
+            session.query(DagRun).delete()
+            session.query(TI).delete()
+
+    def _assert_task_ids_match_states(self, task_ids_to_states):
+        """Helper that asserts task instances with a given id are in a given state"""
+        tis = self.dr.get_task_instances()
+        for ti in tis:
+            try:
+                expected_state = task_ids_to_states[ti.task_id]
+            except KeyError:
+                raise ValueError(f'Invalid task id {ti.task_id} found!')
+            else:
+                self.assertEqual(
+                    ti.state,
+                    expected_state,
+                    f"Task {ti.task_id} has state {ti.state} instead of expected {expected_state}",
+                )
+
+    def test_no_target_time(self):
+        """Check if DateTimeBranchOperator raises exception on missing target"""
+        with self.assertRaises(AirflowException):
+            DateTimeBranchOperator(
+                task_id='datetime_branch',
+                follow_task_ids_if_true='branch_1',
+                follow_task_ids_if_false='branch_2',
+                target_upper=None,
+                target_lower=None,
+                dag=self.dag,
+            )
+
+    @freezegun.freeze_time("2020-07-07 10:54:05")
+    def test_datetime_branch_operator_falls_within_range(self):
+        """Check DateTimeBranchOperator branch operation"""
+        for target_lower, target_upper in self.targets:
+            with self.subTest(target_lower=target_lower, target_upper=target_upper):
+                print(f"{target_lower}, {target_upper}")
+                self.branch_op.target_lower = target_lower
+                self.branch_op.target_upper = target_upper
+                self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+
+                self._assert_task_ids_match_states(
+                    {
+                        'datetime_branch': State.SUCCESS,
+                        'branch_1': State.NONE,
+                        'branch_2': State.SKIPPED,
+                    }
+                )
+
+    def test_datetime_branch_operator_falls_outside_range(self):
+        """Check DateTimeBranchOperator branch operation"""
+        dates = [
+            datetime.datetime(2020, 7, 7, 12, 0, 0, tzinfo=datetime.timezone.utc),
+            datetime.datetime(2020, 6, 7, 12, 0, 0, tzinfo=datetime.timezone.utc),
+        ]
+
+        for target_lower, target_upper in self.targets:
+            with self.subTest(target_lower=target_lower, target_upper=target_upper):
+                self.branch_op.target_lower = target_lower
+                self.branch_op.target_upper = target_upper
+
+                for date in dates:
+                    with freezegun.freeze_time(date):
+                        self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+
+                        self._assert_task_ids_match_states(
+                            {
+                                'datetime_branch': State.SUCCESS,
+                                'branch_1': State.SKIPPED,
+                                'branch_2': State.NONE,
+                            }
+                        )
+
+    @freezegun.freeze_time("2020-07-07 10:54:05")
+    def test_datetime_branch_operator_upper_comparison_within_range(self):
+        """Check DateTimeBranchOperator branch operation"""
+        for _, target_upper in self.targets:
+            with self.subTest(target_upper=target_upper):
+                self.branch_op.target_upper = target_upper
+                self.branch_op.target_lower = None
+
+                self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+
+                self._assert_task_ids_match_states(
+                    {
+                        'datetime_branch': State.SUCCESS,
+                        'branch_1': State.NONE,
+                        'branch_2': State.SKIPPED,
+                    }
+                )
+
+    @freezegun.freeze_time("2020-07-07 10:54:05")
+    def test_datetime_branch_operator_lower_comparison_within_range(self):
+        """Check DateTimeBranchOperator branch operation"""
+        for target_lower, _ in self.targets:
+            with self.subTest(target_lower=target_lower):
+                self.branch_op.target_lower = target_lower
+                self.branch_op.target_upper = None
+
+                self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+
+                self._assert_task_ids_match_states(
+                    {
+                        'datetime_branch': State.SUCCESS,
+                        'branch_1': State.NONE,
+                        'branch_2': State.SKIPPED,
+                    }
+                )
+
+    @freezegun.freeze_time("2020-07-07 12:00:00")
+    def test_datetime_branch_operator_upper_comparison_outside_range(self):
+        """Check DateTimeBranchOperator branch operation"""
+        for _, target_upper in self.targets:
+            with self.subTest(target_upper=target_upper):
+                self.branch_op.target_upper = target_upper
+                self.branch_op.target_lower = None
+
+                self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+
+                self._assert_task_ids_match_states(
+                    {
+                        'datetime_branch': State.SUCCESS,
+                        'branch_1': State.SKIPPED,
+                        'branch_2': State.NONE,
+                    }
+                )
+
+    @freezegun.freeze_time("2020-07-07 09:00:00")
+    def test_datetime_branch_operator_lower_comparison_outside_range(self):
+        """Check DateTimeBranchOperator branch operation"""
+        for target_lower, _ in self.targets:
+            with self.subTest(target_lower=target_lower):
+                self.branch_op.target_lower = target_lower
+                self.branch_op.target_upper = None
+
+                self.branch_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
+
+                self._assert_task_ids_match_states(
+                    {
+                        'datetime_branch': State.SUCCESS,
+                        'branch_1': State.SKIPPED,
+                        'branch_2': State.NONE,
+                    }
+                )


Code review comments:
diff_hunk: @@ -0,0 +1,101 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import datetime
+from typing import Dict, Iterable, Union
+
+from airflow.exceptions import AirflowException
+from airflow.operators.branch_operator import BaseBranchOperator
+from airflow.utils import timezone
+from airflow.utils.decorators import apply_defaults
+
+
+class DateTimeBranchOperator(BaseBranchOperator):
+    """
+    Branches into one of two lists of tasks depending on the current datetime.
+
+    True branch will be returned when `datetime.datetime.now()` falls below
+    `target_upper` and above `target_lower`.
path: airflow/operators/datetime_branch.py
text: @user1:
What's the behaviour when backfilling or catching-up ?

@user1:
How about adding something like:

https://github.com/apache/airflow/blob/e363ac1f33cd82188fb0923791f5cb747629d501/airflow/operators/weekday.py#L86-L89

cc @user2 

@author:
Good call! Behavior when backfilling would be to still use system date, which is probably not what a user would expect. I added a `use_task_execution_date` argument very similar to the one you linked, with some minor adaptation since we are talking about dates instead of days of the week.

diff_hunk: @@ -0,0 +1,101 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import datetime
+from typing import Dict, Iterable, Union
+
+from airflow.exceptions import AirflowException
+from airflow.operators.branch_operator import BaseBranchOperator
+from airflow.utils import timezone
+from airflow.utils.decorators import apply_defaults
+
+
+class DateTimeBranchOperator(BaseBranchOperator):
+    """
+    Branches into one of two lists of tasks depending on the current datetime.
+
+    True branch will be returned when `datetime.datetime.now()` falls below
+    `target_upper` and above `target_lower`.
+
+    :param follow_task_ids_if_true: task id or task ids to follow if
+        `datetime.datetime.now()` falls above target_lower and below `target_upper`.
path: airflow/operators/datetime_branch.py
text: @user1:
```suggestion
        ``datetime.datetime.now()`` falls above target_lower and below ``target_upper``.
```

diff_hunk: @@ -0,0 +1,101 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import datetime
+from typing import Dict, Iterable, Union
+
+from airflow.exceptions import AirflowException
+from airflow.operators.branch_operator import BaseBranchOperator
+from airflow.utils import timezone
+from airflow.utils.decorators import apply_defaults
+
+
+class DateTimeBranchOperator(BaseBranchOperator):
+    """
+    Branches into one of two lists of tasks depending on the current datetime.
+
+    True branch will be returned when `datetime.datetime.now()` falls below
+    `target_upper` and above `target_lower`.
+
+    :param follow_task_ids_if_true: task id or task ids to follow if
+        `datetime.datetime.now()` falls above target_lower and below `target_upper`.
+    :type follow_task_ids_if_true: str or list[str]
+    :param follow_task_ids_if_false: task id or task ids to follow if
+        `datetime.datetime.now()` falls below target_lower or above `target_upper`.
path: airflow/operators/datetime_branch.py
text: @user1:
```suggestion
        ``datetime.datetime.now()`` falls below target_lower or above ``target_upper``.
```