Refactor datetime_range_filter_factory: coalesce only start_date and end_date filters

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

airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py

@@ -17,12 +17,11 @@
 
 from __future__ import annotations
 
-from collections.abc import Sequence
-from typing import Annotated, Any, Literal, cast
+from typing import Annotated, Literal, cast
 
 import structlog
 from fastapi import Depends, HTTPException, Query, status
-from sqlalchemy import func, or_, select
+from sqlalchemy import or_, select
 from sqlalchemy.orm import joinedload
 from sqlalchemy.sql.selectable import Select
 
@@ -50,7 +49,6 @@
     float_range_filter_factory,
 )
 from airflow.api_fastapi.common.router import AirflowRouter
-from airflow.api_fastapi.core_api.base import OrmClause
 from airflow.api_fastapi.core_api.datamodels.common import BulkBody, BulkResponse
 from airflow.api_fastapi.core_api.datamodels.task_instances import (
     BulkTaskInstanceBody,
@@ -196,26 +194,21 @@ def get_mapped_task_instances(
             error_message = f"Task id {task_id} is not mapped"
             raise HTTPException(status.HTTP_404_NOT_FOUND, error_message)
 
-    filters = [
-        run_after_range,
-        logical_date_range,
-        start_date_range,
-        end_date_range,
-        update_at_range,
-        duration_range,
-        state,
-        pool,
-        queue,
-        executor,
-        version_number,
-    ]
-    if state.value is not None and None in state.value:
-        current_time = func.now()
-        for f in (start_date_range, end_date_range):
-            f.attribute = func.coalesce(f.attribute, current_time)
     task_instance_select, total_entries = paginated_select(
         statement=query,
-        filters=cast("Sequence[OrmClause[Any]]", filters),
+        filters=[
+            run_after_range,
+            logical_date_range,
+            start_date_range,
+            end_date_range,
+            update_at_range,
+            duration_range,
+            state,
+            pool,
+            queue,
+            executor,
+            version_number,
+        ],
         order_by=order_by,
         offset=offset,
         limit=limit,
@@ -467,26 +460,24 @@ def get_task_instances(
             )
         query = query.where(TI.run_id == dag_run_id)
 
-    filters = [
-        run_after_range,
-        logical_date_range,
-        start_date_range,
-        end_date_range,
-        update_at_range,
-        duration_range,
-        state,
-        pool,
-        queue,
-        executor,
-        version_number,
-    ]
-    if state.value is not None and None in state.value:
-        current_time = func.now()
-        for f in (start_date_range, end_date_range):
-            f.attribute = func.coalesce(f.attribute, current_time)
     task_instance_select, total_entries = paginated_select(
         statement=query,
-        filters=cast("Sequence[OrmClause[Any]]", filters),
+        filters=[
+            run_after_range,
+            logical_date_range,
+            start_date_range,
+            end_date_range,
+            update_at_range,
+            duration_range,
+            state,
+            pool,
+            queue,
+            executor,
+            task_id,
+            task_display_name_pattern,
+            version_number,
+            readable_ti_filter,
+        ],
         order_by=order_by,
         offset=offset,
         limit=limit,