Add DateTimeSensor

[22quinn]

Related to #9609

A major advantage of this new sensor is idempotence for the target_time.

It handles some cases for which TimeSensor and TimeDeltaSensor are not suited.

Example 1) If a task needs to wait for 11am on each execution_date. Using TimeSensor or TimeDeltaSensor, all backfill tasks started at 1am have to wait for 10 hours. This is unnecessary, e.g. a backfill task with {{ ds }} = '1970-01-01' does not need to wait because 1970-01-01T11:00:00 has already passed.

Example 2) If a DAG is scheduled to run at 23:00 daily, but one of the tasks is required to run at 01:00 next day, using TimeSensor will return True immediately because 23:00 > 01:00. Instead, we can use DateTimeSensor:

        DateTimeSensor(
            task_id='wait_for_0100',
            target_time='{{ next_execution_date.tomorrow().replace(hour=1) }}',
            dag=dag
        )

On another topic - I think TimeSensor, TimeDeltaSensor, and this DateTimeSensor should probably be put in the same file sensors/time.py. What does everyone think? But that will be a breaking change. Will have to do it in a separate PR.

---

Make sure to mark the boxes below before creating PR: [x]

• Description above provides context of the change
• Unit tests coverage for changes (not needed for documentation changes)
• Target Github ISSUE in description if exists
• Commits follow "How to write a good git commit message"
• Relevant documentation is updated including usage instructions.
• I will engage committers as explained in Contribution Workflow Example.

---

In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.
Read the Pull Request Guidelines for more information.

[turbaszek]

Would you mind adding a test for poke method?

[22quinn]

Added and rebased

[22quinn]

I'm wondering if it should be changed to>=? Same applies to TimeSensor and TimeDeltaSensor. The impact of changing to >= is that the task can succeed one poke interval earlier in some cases.

[turbaszek]

I would say that > is safer

[22quinn]

I was just thinking >= is more mathematically correct. But now I realize it's extremely rare to have = because utcnow() has microsecond precision. So nvm

[turbaszek]

@zikun before we merge, I think it would be helpful to add some info from PR description to docstring:

    A major advantage of this sensor is idempotence for the ``target_time``.
    It handles some cases for which ``TimeSensor`` and ``TimeDeltaSensor`` are not suited.

    **Example**: 1 ::
        If a task needs to wait for 11am on each ``execution_date``. Using
        ``TimeSensor`` or ``TimeDeltaSensor``, all backfill tasks started at 1am have to wait for 10 hours.
        This is unnecessary, e.g. a backfill task with ``{{ ds }} = '1970-01-01'`` does not need to wait
        because ``1970-01-01T11:00:00`` has already passed.

    **Example**: 2 ::
        If a DAG is scheduled to run at 23:00 daily, but one of the tasks is required
        to run at 01:00 next day, using ``TimeSensor`` will return ``True`` immediately because 23:00 > 01:00.
        Instead, we can use ``DateTimeSensor``:

        .. code-block:: python
            DateTimeSensor(
                task_id='wait_for_0100',
                target_time='{{ next_execution_date.tomorrow().replace(hour=1) }}',
                dag=dag
            )

Hope this is ok formatted rst but I'm not sure

[22quinn]

I tried on rst online viewer. It should work.