Fix Task Instance “No Status” Filter (#51880)

* Support no_status alias in TaskInstance state filter for REST API

* Allow 'no_status' state filter and include no_status in valid state list; skip date filters when filtering for null state

* Fix NULL-state filtering in get_mapped_task_instances by coalescing date fields

* Refactor datetime_range_filter_factory: coalesce only start_date and end_date filters

* Add a test

Author: sunank200

airflow-core/src/airflow/api_fastapi/common/parameters.py

@@ -37,7 +37,7 @@
 from fastapi import Depends, HTTPException, Query, status
 from pendulum.parsing.exceptions import ParserError
 from pydantic import AfterValidator, BaseModel, NonNegativeInt
-from sqlalchemy import Column, and_, case, or_
+from sqlalchemy import Column, and_, case, func, or_
 from sqlalchemy.inspection import inspect
 
 from airflow.api_fastapi.core_api.base import OrmClause
@@ -493,9 +493,12 @@ def depends_datetime(
         lower_bound: datetime | None = Query(alias=f"{filter_name}_gte", default=None),
         upper_bound: datetime | None = Query(alias=f"{filter_name}_lte", default=None),
     ) -> RangeFilter:
+        attr = getattr(model, attribute_name or filter_name)
+        if filter_name in ("start_date", "end_date"):
+            attr = func.coalesce(attr, func.now())
         return RangeFilter(
             Range(lower_bound=lower_bound, upper_bound=upper_bound),
-            getattr(model, attribute_name or filter_name),
+            attr,
         )
 
     return depends_datetime
@@ -610,7 +613,7 @@ def _transform_ti_states(states: list[str] | None) -> list[TaskInstanceState | N
         return None
 
     try:
-        return [None if s in ("none", None) else TaskInstanceState(s) for s in states]
+        return [None if s in ("no_status", "none", None) else TaskInstanceState(s) for s in states]
     except ValueError:
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py

@@ -964,6 +964,19 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
                 3,
                 id="test state filter",
             ),
+            pytest.param(
+                [
+                    {"state": State.RUNNING},
+                    {"state": State.QUEUED},
+                    {"state": State.SUCCESS},
+                    {"state": State.NONE},
+                ],
+                False,
+                ("/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances"),
+                {"state": ["no_status"]},
+                1,
+                id="test no_status state filter",
+            ),
             pytest.param(
                 [
                     {"state": State.NONE},
@@ -977,6 +990,14 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
                 4,
                 id="test null states with no filter",
             ),
+            pytest.param(
+                [{"start_date": None, "end_date": None}],
+                True,
+                "/dags/example_python_operator/dagRuns/TEST_DAG_RUN_ID/taskInstances",
+                {"start_date_gte": DEFAULT_DATETIME_STR_1},
+                1,
+                id="test start_date coalesce with null",
+            ),
             pytest.param(
                 [
                     {"pool": "test_pool_1"},