feat(hitl): add filters and order by to get_hitl_details public API

* filter: dag_id, dag_run_id, ti_state, response_recevied, user_id, subject, body
* sort_by: dag_id, run_id

Author: Lee-W

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

@@ -52,6 +52,7 @@
 from airflow.models.dag_favorite import DagFavorite
 from airflow.models.dag_version import DagVersion
 from airflow.models.dagrun import DagRun
+from airflow.models.hitl import HITLDetail
 from airflow.models.pool import Pool
 from airflow.models.taskinstance import TaskInstance
 from airflow.models.variable import Variable
@@ -750,3 +751,27 @@ def _optional_boolean(value: bool | None) -> bool | None:
 QueryConnectionIdPatternSearch = Annotated[
     _SearchParam, Depends(search_param_factory(Connection.conn_id, "connection_id_pattern"))
 ]
+
+# Human in the loop
+QueryDagRunIdFilter = Annotated[
+    FilterParam[list[str]],
+    Depends(filter_param_factory(DagRun.id, list[str], filter_name="dag_run_id")),
+]
+QueryHITLDetailResponseReceivedFilter = Annotated[
+    FilterParam[bool | None],
+    Depends(
+        filter_param_factory(
+            HITLDetail.response_received, bool | None, filter_name="hitl_detail_response_received"
+        )
+    ),
+]
+QueryHITLDetailUserIdFilter = Annotated[
+    FilterParam[list[str]],
+    Depends(filter_param_factory(HITLDetail.user_id, list[str], filter_name="hitl_detail_user_id")),
+]
+QueryHITLDetailSubjectSearch = Annotated[
+    _SearchParam, Depends(search_param_factory(HITLDetail.subject, "hitl_detail_subject_search"))
+]
+QueryHITLDetailBodySearch = Annotated[
+    _SearchParam, Depends(search_param_factory(HITLDetail.body, "hitl_detail_body_search"))
+]

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -7399,6 +7399,100 @@ paths:
       summary: Get Hitl Details
       description: Get Human-in-the-loop details.
       operationId: get_hitl_details
+      security:
+      - OAuth2PasswordBearer: []
+      parameters:
+      - name: limit
+        in: query
+        required: false
+        schema:
+          type: integer
+          minimum: 0
+          default: 50
+          title: Limit
+      - name: offset
+        in: query
+        required: false
+        schema:
+          type: integer
+          minimum: 0
+          default: 0
+          title: Offset
+      - name: order_by
+        in: query
+        required: false
+        schema:
+          type: string
+          default: map_index
+          title: Order By
+      - name: dag_id_pattern
+        in: query
+        required: false
+        schema:
+          anyOf:
+          - type: string
+          - type: 'null'
+          description: "SQL LIKE expression \u2014 use `%` / `_` wildcards (e.g. `%customer_%`).\
+            \ Regular expressions are **not** supported."
+          title: Dag Id Pattern
+        description: "SQL LIKE expression \u2014 use `%` / `_` wildcards (e.g. `%customer_%`).\
+          \ Regular expressions are **not** supported."
+      - name: dag_run_id
+        in: query
+        required: false
+        schema:
+          type: array
+          items:
+            type: string
+          title: Dag Run Id
+      - name: state
+        in: query
+        required: false
+        schema:
+          type: array
+          items:
+            type: string
+          title: State
+      - name: hitl_detail_response_received
+        in: query
+        required: false
+        schema:
+          anyOf:
+          - type: boolean
+          - type: 'null'
+          title: Hitl Detail Response Received
+      - name: hitl_detail_user_id
+        in: query
+        required: false
+        schema:
+          type: array
+          items:
+            type: string
+          title: Hitl Detail User Id
+      - name: hitl_detail_subject_search
+        in: query
+        required: false
+        schema:
+          anyOf:
+          - type: string
+          - type: 'null'
+          description: "SQL LIKE expression \u2014 use `%` / `_` wildcards (e.g. `%customer_%`).\
+            \ Regular expressions are **not** supported."
+          title: Hitl Detail Subject Search
+        description: "SQL LIKE expression \u2014 use `%` / `_` wildcards (e.g. `%customer_%`).\
+          \ Regular expressions are **not** supported."
+      - name: hitl_detail_body_search
+        in: query
+        required: false
+        schema:
+          anyOf:
+          - type: string
+          - type: 'null'
+          description: "SQL LIKE expression \u2014 use `%` / `_` wildcards (e.g. `%customer_%`).\
+            \ Regular expressions are **not** supported."
+          title: Hitl Detail Body Search
+        description: "SQL LIKE expression \u2014 use `%` / `_` wildcards (e.g. `%customer_%`).\
+          \ Regular expressions are **not** supported."
       responses:
         '200':
           description: Successful Response
@@ -7407,19 +7501,23 @@ paths:
               schema:
                 $ref: '#/components/schemas/HITLDetailCollection'
         '401':
-          description: Unauthorized
           content:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Unauthorized
         '403':
-          description: Forbidden
           content:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPExceptionResponse'
-      security:
-      - OAuth2PasswordBearer: []
+          description: Forbidden
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
   /api/v2/monitor/health:
     get:
       tags:
@@ -9891,9 +9989,8 @@ components:
       description: Serializer for Plugin FastAPI root middleware responses.
     HITLDetail:
       properties:
-        ti_id:
-          type: string
-          title: Ti Id
+        task_instance:
+          $ref: '#/components/schemas/TaskInstanceResponse'
         options:
           items:
             type: string
@@ -9950,7 +10047,7 @@ components:
           default: false
       type: object
       required:
-      - ti_id
+      - task_instance
       - options
       - subject
       title: HITLDetail

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

@@ -16,13 +16,27 @@
 # under the License.
 from __future__ import annotations
 
+from typing import Annotated
+
 import structlog
 from fastapi import Depends, HTTPException, status
 from sqlalchemy import select
 from sqlalchemy.orm import joinedload
 
 from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
 from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
+from airflow.api_fastapi.common.parameters import (
+    QueryDagIdPatternSearch,
+    QueryDagRunIdFilter,
+    QueryHITLDetailBodySearch,
+    QueryHITLDetailResponseReceivedFilter,
+    QueryHITLDetailSubjectSearch,
+    QueryHITLDetailUserIdFilter,
+    QueryLimit,
+    QueryOffset,
+    QueryTIStateFilter,
+    SortParam,
+)
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.core_api.datamodels.hitl import (
     HITLDetail,
@@ -262,16 +276,55 @@ def get_mapped_ti_hitl_detail(
     dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.TASK_INSTANCE))],
 )
 def get_hitl_details(
-    readable_ti_filter: ReadableTIFilterDep,
+    limit: QueryLimit,
+    offset: QueryOffset,
+    order_by: Annotated[
+        SortParam,
+        Depends(
+            SortParam(
+                ["dag_id", "run_id"],
+                TI,
+            ).dynamic_depends(default="map_index"),
+        ),
+    ],
     session: SessionDep,
+    # ti related filter
+    readable_ti_filter: ReadableTIFilterDep,
+    dag_id_pattern: QueryDagIdPatternSearch,
+    dag_run_id: QueryDagRunIdFilter,
+    ti_state: QueryTIStateFilter,
+    # hitl detail related filter
+    response_received: QueryHITLDetailResponseReceivedFilter,
+    user_id: QueryHITLDetailUserIdFilter,
+    subject_patten: QueryHITLDetailSubjectSearch,
+    body_patten: QueryHITLDetailBodySearch,
 ) -> HITLDetailCollection:
     """Get Human-in-the-loop details."""
-    query = select(HITLDetailModel).join(TI, HITLDetailModel.ti_id == TI.id)
+    query = (
+        select(HITLDetailModel)
+        .join(TI, HITLDetailModel.ti_id == TI.id)
+        .options(joinedload(HITLDetailModel.task_instance))
+    )
     hitl_detail_select, total_entries = paginated_select(
         statement=query,
-        filters=[readable_ti_filter],
+        filters=[
+            # ti related filter
+            readable_ti_filter,
+            dag_id_pattern,
+            dag_run_id,
+            ti_state,
+            # hitl detail related filter
+            response_received,
+            user_id,
+            subject_patten,
+            body_patten,
+        ],
+        offset=offset,
+        limit=limit,
+        order_by=order_by,
         session=session,
     )
+
     hitl_details = session.scalars(hitl_detail_select)
     return HITLDetailCollection(
         hitl_details=hitl_details,

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

@@ -73,3 +73,7 @@ class HITLDetail(Base):
     @hybrid_property
     def response_received(self) -> bool:
         return self.response_at is not None
+
+    @response_received.expression  # type: ignore[no-redef]
+    def response_received(cls):
+        return cls.response_at.is_not(None)

airflow-ctl/src/airflowctl/api/datamodels/generated.py

@@ -573,34 +573,6 @@ class FastAPIRootMiddlewareResponse(BaseModel):
     name: Annotated[str, Field(title="Name")]
 
 
-class HITLDetail(BaseModel):
-    """
-    Schema for Human-in-the-loop detail.
-    """
-
-    ti_id: Annotated[str, Field(title="Ti Id")]
-    options: Annotated[list[str], Field(title="Options")]
-    subject: Annotated[str, Field(title="Subject")]
-    body: Annotated[str | None, Field(title="Body")] = None
-    defaults: Annotated[list[str] | None, Field(title="Defaults")] = None
-    multiple: Annotated[bool | None, Field(title="Multiple")] = False
-    params: Annotated[dict[str, Any] | None, Field(title="Params")] = None
-    user_id: Annotated[str | None, Field(title="User Id")] = None
-    response_at: Annotated[datetime | None, Field(title="Response At")] = None
-    chosen_options: Annotated[list[str] | None, Field(title="Chosen Options")] = None
-    params_input: Annotated[dict[str, Any] | None, Field(title="Params Input")] = None
-    response_received: Annotated[bool | None, Field(title="Response Received")] = False
-
-
-class HITLDetailCollection(BaseModel):
-    """
-    Schema for a collection of Human-in-the-loop details.
-    """
-
-    hitl_details: Annotated[list[HITLDetail], Field(title="Hitl Details")]
-    total_entries: Annotated[int, Field(title="Total Entries")]
-
-
 class HITLDetailResponse(BaseModel):
     """
     Response of updating a Human-in-the-loop detail.
@@ -1829,6 +1801,34 @@ class DagStatsCollectionResponse(BaseModel):
     total_entries: Annotated[int, Field(title="Total Entries")]
 
 
+class HITLDetail(BaseModel):
+    """
+    Schema for Human-in-the-loop detail.
+    """
+
+    task_instance: TaskInstanceResponse
+    options: Annotated[list[str], Field(title="Options")]
+    subject: Annotated[str, Field(title="Subject")]
+    body: Annotated[str | None, Field(title="Body")] = None
+    defaults: Annotated[list[str] | None, Field(title="Defaults")] = None
+    multiple: Annotated[bool | None, Field(title="Multiple")] = False
+    params: Annotated[dict[str, Any] | None, Field(title="Params")] = None
+    user_id: Annotated[str | None, Field(title="User Id")] = None
+    response_at: Annotated[datetime | None, Field(title="Response At")] = None
+    chosen_options: Annotated[list[str] | None, Field(title="Chosen Options")] = None
+    params_input: Annotated[dict[str, Any] | None, Field(title="Params Input")] = None
+    response_received: Annotated[bool | None, Field(title="Response Received")] = False
+
+
+class HITLDetailCollection(BaseModel):
+    """
+    Schema for a collection of Human-in-the-loop details.
+    """
+
+    hitl_details: Annotated[list[HITLDetail], Field(title="Hitl Details")]
+    total_entries: Annotated[int, Field(title="Total Entries")]
+
+
 class PluginCollectionResponse(BaseModel):
     """
     Plugin Collection serializer.

scripts/ci/pre_commit/check_ti_vs_tis_attributes.py

@@ -56,6 +56,7 @@ def compare_attributes(path1, path2):
         "triggerer_job",
         "note",
         "rendered_task_instance_fields",
+        "hitl_detail",
         # Storing last heartbeat for historic TIs is not interesting/useful
         "last_heartbeat_at",
         "id",