Fetch asset event source dag run (#59090)

Author: uranusjr

airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py

@@ -20,8 +20,10 @@
 import logging
 from typing import Annotated
 
-from fastapi import APIRouter, HTTPException, Query, status
+from cadwyn import VersionedAPIRouter
+from fastapi import HTTPException, Query, status
 from sqlalchemy import func, select
+from sqlalchemy.exc import NoResultFound
 
 from airflow.api.common.trigger_dag import trigger_dag
 from airflow.api_fastapi.common.dagbag import DagBagDep, get_dag_for_run
@@ -36,19 +38,54 @@
 from airflow.utils.state import DagRunState
 from airflow.utils.types import DagRunTriggeredByType
 
-router = APIRouter()
-
+router = VersionedAPIRouter()
 
 log = logging.getLogger(__name__)
 
 
+@router.only_exists_in_older_versions
+@router.get("/{dag_id}/previous")
+def get_previous_dagrun_compat(
+    dag_id: str,
+    logical_date: UtcDateTime,
+    session: SessionDep,
+    state: DagRunState | None = None,
+):
+    """
+    Redirect old previous dag run request to the new endpoint.
+
+    This endpoint must be put before ``get_dag_run`` so not to be shadowed.
+    Newer client versions would not see this endpoint, and be routed to
+    ``get_dag_run`` below instead.
+    """
+    return get_previous_dagrun(dag_id, logical_date, session, state)
+
+
+@router.get(
+    "/{dag_id}/{run_id}",
+    responses={status.HTTP_404_NOT_FOUND: {"description": "Dag run not found"}},
+)
+def get_dag_run(dag_id: str, run_id: str, session: SessionDep) -> DagRun:
+    """Get detail of a Dag run."""
+    dr = session.scalar(select(DagRunModel).where(DagRunModel.dag_id == dag_id, DagRunModel.run_id == run_id))
+    if dr is None:
+        raise HTTPException(
+            status.HTTP_404_NOT_FOUND,
+            detail={
+                "reason": "not_found",
+                "message": f"Dag run with dag_id '{dag_id}' and run_id '{run_id}' was not found",
+            },
+        )
+    return DagRun.model_validate(dr)
+
+
 @router.post(
     "/{dag_id}/{run_id}",
     status_code=status.HTTP_204_NO_CONTENT,
     responses={
-        status.HTTP_400_BAD_REQUEST: {"description": "DAG has import errors and cannot be triggered"},
-        status.HTTP_404_NOT_FOUND: {"description": "DAG not found for the given dag_id"},
-        status.HTTP_409_CONFLICT: {"description": "DAG Run already exists for the given dag_id"},
+        status.HTTP_400_BAD_REQUEST: {"description": "Dag has import errors and cannot be triggered"},
+        status.HTTP_404_NOT_FOUND: {"description": "Dag not found for the given dag_id"},
+        status.HTTP_409_CONFLICT: {"description": "Dag run already exists for the given dag_id"},
         HTTP_422_UNPROCESSABLE_CONTENT: {"description": "Invalid payload"},
     },
 )
@@ -57,21 +94,21 @@ def trigger_dag_run(
     run_id: str,
     payload: TriggerDAGRunPayload,
     session: SessionDep,
-):
-    """Trigger a DAG Run."""
+) -> None:
+    """Trigger a Dag run."""
     dm = session.scalar(select(DagModel).where(~DagModel.is_stale, DagModel.dag_id == dag_id).limit(1))
     if not dm:
         raise HTTPException(
             status.HTTP_404_NOT_FOUND,
-            detail={"reason": "not_found", "message": f"DAG with dag_id: '{dag_id}' not found"},
+            detail={"reason": "not_found", "message": f"Dag with dag_id: '{dag_id}' not found"},
         )
 
     if dm.has_import_errors:
         raise HTTPException(
             status.HTTP_400_BAD_REQUEST,
             detail={
                 "reason": "import_errors",
-                "message": f"DAG with dag_id: '{dag_id}' has import errors and cannot be triggered",
+                "message": f"Dag with dag_id '{dag_id}' has import errors and cannot be triggered",
             },
         )
 
@@ -90,7 +127,7 @@ def trigger_dag_run(
             status.HTTP_409_CONFLICT,
             detail={
                 "reason": "already_exists",
-                "message": f"A DAG Run already exists for DAG {dag_id} with run id {run_id}",
+                "message": f"A run already exists for Dag '{dag_id}' with run_id '{run_id}'",
             },
         )
 
@@ -99,8 +136,8 @@ def trigger_dag_run(
     "/{dag_id}/{run_id}/clear",
     status_code=status.HTTP_204_NO_CONTENT,
     responses={
-        status.HTTP_400_BAD_REQUEST: {"description": "DAG has import errors and cannot be triggered"},
-        status.HTTP_404_NOT_FOUND: {"description": "DAG not found for the given dag_id"},
+        status.HTTP_400_BAD_REQUEST: {"description": "Dag has import errors and cannot be triggered"},
+        status.HTTP_404_NOT_FOUND: {"description": "Dag not found for the given dag_id"},
         HTTP_422_UNPROCESSABLE_CONTENT: {"description": "Invalid payload"},
     },
 )
@@ -109,21 +146,21 @@ def clear_dag_run(
     run_id: str,
     session: SessionDep,
     dag_bag: DagBagDep,
-):
-    """Clear a DAG Run."""
+) -> None:
+    """Clear a Dag run."""
     dm = session.scalar(select(DagModel).where(~DagModel.is_stale, DagModel.dag_id == dag_id).limit(1))
     if not dm:
         raise HTTPException(
             status.HTTP_404_NOT_FOUND,
-            detail={"reason": "not_found", "message": f"DAG with dag_id: '{dag_id}' not found"},
+            detail={"reason": "not_found", "message": f"Dag with dag_id: '{dag_id}' not found"},
         )
 
     if dm.has_import_errors:
         raise HTTPException(
             status.HTTP_400_BAD_REQUEST,
             detail={
                 "reason": "import_errors",
-                "message": f"DAG with dag_id: '{dag_id}' has import errors and cannot be triggered",
+                "message": f"Dag with dag_id '{dag_id}' has import errors and cannot be triggered",
             },
         )
 
@@ -133,7 +170,7 @@ def clear_dag_run(
     if dag_run is None:
         raise HTTPException(
             status.HTTP_404_NOT_FOUND,
-            detail={"reason": "not_found", "message": f"DAG run with run_id: '{run_id}' not found"},
+            detail={"reason": "not_found", "message": f"Dag run with run_id: '{run_id}' not found"},
         )
     dag = get_dag_for_run(dag_bag, dag_run=dag_run, session=session)
 
@@ -142,29 +179,27 @@ def clear_dag_run(
 
 @router.get(
     "/{dag_id}/{run_id}/state",
-    responses={
-        status.HTTP_404_NOT_FOUND: {"description": "DAG not found for the given dag_id"},
-    },
+    responses={status.HTTP_404_NOT_FOUND: {"description": "Dag run not found"}},
 )
 def get_dagrun_state(
     dag_id: str,
     run_id: str,
     session: SessionDep,
 ) -> DagRunStateResponse:
-    """Get a DAG Run State."""
-    dag_run = session.scalar(
-        select(DagRunModel).where(DagRunModel.dag_id == dag_id, DagRunModel.run_id == run_id)
-    )
-    if dag_run is None:
+    """Get a Dag run State."""
+    try:
+        state = session.scalars(
+            select(DagRunModel.state).where(DagRunModel.dag_id == dag_id, DagRunModel.run_id == run_id)
+        ).one()
+    except NoResultFound:
         raise HTTPException(
             status.HTTP_404_NOT_FOUND,
             detail={
                 "reason": "not_found",
-                "message": f"The DagRun with dag_id: `{dag_id}` and run_id: `{run_id}` was not found",
+                "message": f"Dag run with dag_id '{dag_id}' and run_id '{run_id}' was not found",
             },
         )
-
-    return DagRunStateResponse(state=dag_run.state)
+    return DagRunStateResponse(state=state)
 
 
 @router.get("/count", status_code=status.HTTP_200_OK)
@@ -175,46 +210,33 @@ def get_dr_count(
     run_ids: Annotated[list[str] | None, Query()] = None,
     states: Annotated[list[str] | None, Query()] = None,
 ) -> int:
-    """Get the count of DAG runs matching the given criteria."""
-    query = select(func.count()).select_from(DagRunModel).where(DagRunModel.dag_id == dag_id)
-
+    """Get the count of Dag runs matching the given criteria."""
+    stmt = select(func.count()).select_from(DagRunModel).where(DagRunModel.dag_id == dag_id)
     if logical_dates:
-        query = query.where(DagRunModel.logical_date.in_(logical_dates))
-
+        stmt = stmt.where(DagRunModel.logical_date.in_(logical_dates))
     if run_ids:
-        query = query.where(DagRunModel.run_id.in_(run_ids))
-
+        stmt = stmt.where(DagRunModel.run_id.in_(run_ids))
     if states:
-        query = query.where(DagRunModel.state.in_(states))
+        stmt = stmt.where(DagRunModel.state.in_(states))
+    return session.scalar(stmt) or 0
 
-    count = session.scalar(query)
-    return count or 0
 
-
-@router.get("/{dag_id}/previous", status_code=status.HTTP_200_OK)
+@router.get("/previous", status_code=status.HTTP_200_OK)
 def get_previous_dagrun(
     dag_id: str,
     logical_date: UtcDateTime,
     session: SessionDep,
     state: Annotated[DagRunState | None, Query()] = None,
 ) -> DagRun | None:
-    """Get the previous DAG run before the given logical date, optionally filtered by state."""
-    query = (
+    """Get the previous Dag run before the given logical date, optionally filtered by state."""
+    stmt = (
         select(DagRunModel)
-        .where(
-            DagRunModel.dag_id == dag_id,
-            DagRunModel.logical_date < logical_date,
-        )
+        .where(DagRunModel.dag_id == dag_id, DagRunModel.logical_date < logical_date)
         .order_by(DagRunModel.logical_date.desc())
         .limit(1)
     )
-
     if state:
-        query = query.where(DagRunModel.state == state)
-
-    dag_run = session.scalar(query)
-
-    if not dag_run:
+        stmt = stmt.where(DagRunModel.state == state)
+    if not (dag_run := session.scalar(stmt)):
         return None
-
     return DagRun.model_validate(dag_run)

airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py

@@ -29,9 +29,14 @@
 from airflow.api_fastapi.execution_api.versions.v2025_10_27 import MakeDagRunConfNullable
 from airflow.api_fastapi.execution_api.versions.v2025_11_05 import AddTriggeringUserNameField
 from airflow.api_fastapi.execution_api.versions.v2025_11_07 import AddPartitionKeyField
+from airflow.api_fastapi.execution_api.versions.v2025_12_08 import (
+    AddDagRunDetailEndpoint,
+    MovePreviousRunEndpoint,
+)
 
 bundle = VersionBundle(
     HeadVersion(),
+    Version("2025-12-08", MovePreviousRunEndpoint, AddDagRunDetailEndpoint),
     Version("2025-11-07", AddPartitionKeyField),
     Version("2025-11-05", AddTriggeringUserNameField),
     Version("2025-10-27", MakeDagRunConfNullable),

airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_12_08.py

@@ -0,0 +1,41 @@
+# 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.
+
+from __future__ import annotations
+
+from cadwyn import VersionChange, endpoint
+
+
+class MovePreviousRunEndpoint(VersionChange):
+    """Add new previous-run endpoint and migrate old endpoint."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (
+        endpoint("/dag-runs/previous", ["GET"]).didnt_exist,
+        endpoint("/dag-runs/{dag_id}/previous", ["GET"]).existed,
+    )
+
+
+class AddDagRunDetailEndpoint(VersionChange):
+    """Add dag run detail endpoint."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (
+        endpoint("/dag-runs/{dag_id}/{run_id}", ["GET"]).didnt_exist,
+    )