Custom timetable must return aware datetimes (apache#18420)

airflow/example_dags/plugins/workday.py

@@ -59,7 +59,7 @@ def next_dagrun_info(
                 delta = timedelta(days=1)
             else:  # Last run on Friday -- skip to next Monday.
                 delta = timedelta(days=(7 - last_start_weekday))
-            next_start = DateTime.combine((last_start + delta).date(), Time.min)
+            next_start = DateTime.combine((last_start + delta).date(), Time.min).replace(tzinfo=UTC)
         else:  # This is the first ever run on the regular schedule.
             next_start = restriction.earliest
             if next_start is None:  # No start_date. Don't schedule.

airflow/timetables/base.py

@@ -23,7 +23,11 @@
 
 
 class DataInterval(NamedTuple):
-    """A data interval for a DagRun to operate over."""
+    """A data interval for a DagRun to operate over.
+
+    Both ``start`` and ``end`` **MUST** be "aware", i.e. contain timezone
+    information.
+    """
 
     start: DateTime
     end: DateTime
@@ -44,8 +48,10 @@ class TimeRestriction(NamedTuple):
     These values are generally set on the DAG or task's ``start_date``,
     ``end_date``, and ``catchup`` arguments.
 
-    Both ``earliest`` and ``latest`` are inclusive; a DAG run can happen exactly
-    at either point of time.
+    Both ``earliest`` and ``latest``, if not *None*, are inclusive; a DAG run
+    can happen exactly at either point of time. They are guaranteed to be aware
+    (i.e. contain timezone information) for ``TimeRestriction`` instances
+    created by Airflow.
     """
 
     earliest: Optional[DateTime]
@@ -61,7 +67,10 @@ class DagRunInfo(NamedTuple):
     """
 
     run_after: DateTime
-    """The earliest time this DagRun is created and its tasks scheduled."""
+    """The earliest time this DagRun is created and its tasks scheduled.
+
+    This **MUST** be "aware", i.e. contain timezone information.
+    """
 
     data_interval: DataInterval
     """The data interval this DagRun to operate over."""

airflow/utils/timezone.py

@@ -176,11 +176,11 @@ def parse(string: str, timezone=None) -> DateTime:
 
 
 def coerce_datetime(v: Union[None, dt.datetime, DateTime]) -> Optional[DateTime]:
-    """Convert whatever is passed in to ``pendulum.DateTime``."""
+    """Convert whatever is passed in to an timezone-aware ``pendulum.DateTime``."""
     if v is None:
         return None
-    if isinstance(v, DateTime):
-        return v
     if v.tzinfo is None:
         v = make_aware(v)
+    if isinstance(v, DateTime):
+        return v
     return pendulum.instance(v)

docs/apache-airflow/howto/timetable.rst

@@ -62,6 +62,12 @@ schedule should not include the two weekend days. What we want is:
 
 For simplicity, we will only deal with UTC datetimes in this example.
 
+.. note::
+
+    All datetime values returned by a custom timetable **MUST** be "aware", i.e.
+    contains timezone information. Furthermore, they must use ``pendulum``'s
+    datetime and timezone types.
+
 
 Timetable Registration
 ----------------------