LatestOnlyOperator not working with TaskSDK Dag import.

[gopidesupavan]

Body

Example dag, use the import from airflow.sdk import DAG.

from __future__ import annotations

import datetime

from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.latest_only import LatestOnlyOperator
from airflow.sdk import DAG

with DAG(
    dag_id="latest_only",
    schedule=datetime.timedelta(hours=4),
    start_date=datetime.datetime(2021, 1, 1),
    catchup=False,
    tags=["example2", "example3"],
) as dag:
    latest_only = LatestOnlyOperator(task_id="latest_only")
    task1 = EmptyOperator(task_id="task1")

    latest_only >> task1

We may likely need to port this to TaskSDK

airflow/airflow-core/src/airflow/models/dag.py

Line 571 in 7bd1aa6

def next_dagrun_info(

@kaxil Any reason some methods are in modes/dag.py ?

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[kaxil]

Let me verify.

@kaxil Any reason some methods are in modes/dag.py ?

We have kept the methods needed by the Scheduler in airflow.models.dag as they aren't needed for parsing or execution

[kaxil]

@uranusjr @ashb Thoughts on the following patch to make this working? An assumption here is DataInterval won't be None!

diff --git a/providers/standard/src/airflow/providers/standard/operators/latest_only.py b/providers/standard/src/airflow/providers/standard/operators/latest_only.py
index 930d594756..69d84d153a 100644
--- a/providers/standard/src/airflow/providers/standard/operators/latest_only.py
+++ b/providers/standard/src/airflow/providers/standard/operators/latest_only.py
@@ -25,10 +25,12 @@ from typing import TYPE_CHECKING
 import pendulum
 
 from airflow.providers.standard.operators.branch import BaseBranchOperator
+from airflow.timetables.base import TimeRestriction, DataInterval
 from airflow.utils.types import DagRunType
 
 if TYPE_CHECKING:
-    from airflow.models import DAG, DagRun
+    from airflow.sdk import DAG
+    from airflow.sdk.types import DagRunProtocol as DagRun
 
     try:
         from airflow.sdk.definitions.context import Context
@@ -59,7 +61,13 @@ class LatestOnlyOperator(BaseBranchOperator):
             return list(context["task"].get_direct_relative_ids(upstream=False))
 
         dag: DAG = context["dag"]  # type: ignore[assignment]
-        next_info = dag.next_dagrun_info(dag.get_run_data_interval(dag_run), restricted=False)
+        time_restriction = TimeRestriction(earliest=None, latest=None, catchup=True)
+        next_info = dag.timetable.next_dagrun_info(
+            last_automated_data_interval=DataInterval(
+                start=dag_run.data_interval_start, end=dag_run.data_interval_end
+            ),
+            restriction=time_restriction,
+        )
         now = pendulum.now("UTC")
 
         if next_info is None:

[uranusjr]

Data interval can be None for a very old DagRun. I’ve been wondering if it’d be worthwhile if we do a giant data migration in 3.0 to backfill the values, but there’s not enough motivation.

Since we really only care about the next run anyway, we can probably just default to logical_date if data interval is None. This would only break for very very weird edge cases where you somehow have a run created pre-2.2 but manage to not run it at all until now.

Another (arguably more important) edge case is if the current run (dag_run) is manual and have a null logical date. We should handle this somehow, but I’m not quite sure what makes more sense in this case.