Migrate public endpoint Get Task to FastAPI, with main resynced

airflow/api_connexion/endpoints/task_endpoint.py

@@ -25,12 +25,14 @@
 from airflow.auth.managers.models.resource_details import DagAccessEntity
 from airflow.exceptions import TaskNotFound
 from airflow.utils.airflow_flask_app import get_airflow_app
+from airflow.utils.api_migration import mark_fastapi_migration_done
 
 if TYPE_CHECKING:
     from airflow import DAG
     from airflow.api_connexion.types import APIResponse
 
 
+@mark_fastapi_migration_done
 @security.requires_access_dag("GET", DagAccessEntity.TASK)
 def get_task(*, dag_id: str, task_id: str) -> APIResponse:
     """Get simplified representation of a task."""

airflow/api_fastapi/common/types.py

@@ -16,11 +16,71 @@
 # under the License.
 from __future__ import annotations
 
+import inspect
+from datetime import timedelta
 from typing import Annotated
 
-from pydantic import AfterValidator, AwareDatetime
+from pydantic import AfterValidator, AliasGenerator, AwareDatetime, BaseModel, BeforeValidator, ConfigDict
 
+from airflow.models.mappedoperator import MappedOperator
+from airflow.serialization.serialized_objects import SerializedBaseOperator
 from airflow.utils import timezone
 
 UtcDateTime = Annotated[AwareDatetime, AfterValidator(lambda d: d.astimezone(timezone.utc))]
 """UTCDateTime is a datetime with timezone information"""
+
+
+def _validate_timedelta_field(td: timedelta | None) -> TimeDelta | None:
+    """Validate the execution_timeout property."""
+    if td is None:
+        return None
+    return TimeDelta(
+        days=td.days,
+        seconds=td.seconds,
+        microseconds=td.microseconds,
+    )
+
+
+class TimeDelta(BaseModel):
+    """TimeDelta can be used to interact with datetime.timedelta objects."""
+
+    object_type: str = "TimeDelta"
+    days: int
+    seconds: int
+    microseconds: int
+
+    model_config = ConfigDict(
+        alias_generator=AliasGenerator(
+            serialization_alias=lambda field_name: {
+                "object_type": "__type",
+            }.get(field_name, field_name),
+        )
+    )
+
+
+TimeDeltaWithValidation = Annotated[TimeDelta, BeforeValidator(_validate_timedelta_field)]
+
+
+def get_class_ref(obj) -> dict[str, str | None]:
+    """Return the class_ref dict for obj."""
+    is_mapped_or_serialized = isinstance(obj, (MappedOperator, SerializedBaseOperator))
+
+    module_path = None
+    if is_mapped_or_serialized:
+        module_path = obj._task_module
+    else:
+        module_type = inspect.getmodule(obj)
+        module_path = module_type.__name__ if module_type else None
+
+    class_name = None
+    if is_mapped_or_serialized:
+        class_name = obj._task_type
+    elif obj.__class__ is type:
+        class_name = obj.__name__
+    else:
+        class_name = type(obj).__name__
+
+    return {
+        "module_path": module_path,
+        "class_name": class_name,
+    }

airflow/api_fastapi/core_api/datamodels/tasks.py

@@ -0,0 +1,83 @@
+# 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 collections import abc
+from datetime import datetime
+
+from pydantic import BaseModel, computed_field, field_validator
+
+from airflow.api_fastapi.common.types import TimeDeltaWithValidation
+from airflow.serialization.serialized_objects import encode_priority_weight_strategy
+from airflow.task.priority_strategy import PriorityWeightStrategy
+
+
+class TaskResponse(BaseModel):
+    """Task serializer for responses."""
+
+    task_id: str | None
+    task_display_name: str | None
+    owner: str | None
+    start_date: datetime | None
+    end_date: datetime | None
+    trigger_rule: str | None
+    depends_on_past: bool
+    wait_for_downstream: bool
+    retries: float | None
+    queue: str | None
+    pool: str | None
+    pool_slots: float | None
+    execution_timeout: TimeDeltaWithValidation | None
+    retry_delay: TimeDeltaWithValidation | None
+    retry_exponential_backoff: bool
+    priority_weight: float | None
+    weight_rule: str | None
+    ui_color: str | None
+    ui_fgcolor: str | None
+    template_fields: list[str] | None
+    downstream_task_ids: list[str] | None
+    doc_md: str | None
+    operator_name: str | None
+    params: abc.MutableMapping | None
+    class_ref: dict | None
+    is_mapped: bool | None
+
+    @field_validator("weight_rule", mode="before")
+    @classmethod
+    def validate_weight_rule(cls, wr: str | PriorityWeightStrategy | None) -> str | None:
+        """Validate the weight_rule property."""
+        if wr is None:
+            return None
+        if isinstance(wr, str):
+            return wr
+        return encode_priority_weight_strategy(wr)
+
+    @field_validator("params", mode="before")
+    @classmethod
+    def get_params(cls, params: abc.MutableMapping | None) -> dict | None:
+        """Convert params attribute to dict representation."""
+        if params is None:
+            return None
+        return {param_name: param_val.dump() for param_name, param_val in params.items()}
+
+    # Mypy issue https://github.com/python/mypy/issues/1362
+    @computed_field  # type: ignore[misc]
+    @property
+    def extra_links(self) -> list[str]:
+        """Extract and return extra_links."""
+        return getattr(self, "operator_extra_links", [])

airflow/api_fastapi/core_api/routes/public/__init__.py

@@ -33,6 +33,7 @@
 from airflow.api_fastapi.core_api.routes.public.pools import pools_router
 from airflow.api_fastapi.core_api.routes.public.providers import providers_router
 from airflow.api_fastapi.core_api.routes.public.task_instances import task_instances_router
+from airflow.api_fastapi.core_api.routes.public.tasks import tasks_router
 from airflow.api_fastapi.core_api.routes.public.variables import variables_router
 from airflow.api_fastapi.core_api.routes.public.version import version_router
 from airflow.api_fastapi.core_api.routes.public.xcom import xcom_router
@@ -60,3 +61,5 @@
 public_router.include_router(dag_stats_router)
 public_router.include_router(xcom_router)
 public_router.include_router(assets_router)
+public_router.include_router(plugins_router)
+public_router.include_router(tasks_router)

airflow/api_fastapi/core_api/routes/public/tasks.py

@@ -0,0 +1,56 @@
+# 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 fastapi import HTTPException, Request, status
+
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.common.types import get_class_ref
+from airflow.api_fastapi.core_api.datamodels.tasks import TaskResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+from airflow.models.mappedoperator import MappedOperator
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags/{dag_id}/tasks")
+
+
+@tasks_router.get(
+    "/{task_id}",
+    responses=create_openapi_http_exception_doc(
+        [
+            status.HTTP_400_BAD_REQUEST,
+            status.HTTP_401_UNAUTHORIZED,
+            status.HTTP_403_FORBIDDEN,
+            status.HTTP_404_NOT_FOUND,
+        ]
+    ),
+)
+def get_task(dag_id: str, task_id, request: Request) -> TaskResponse:
+    """Get simplified representation of a task."""
+    dag: DAG = request.app.state.dag_bag.get_dag(dag_id)
+    if not dag:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, f"Dag with id {dag_id} was not found")
+    try:
+        task = dag.get_task(task_id=task_id)
+        task.__dict__.update(
+            {"class_ref": get_class_ref(task), "is_mapped": isinstance(task, MappedOperator)}
+        )
+    except TaskNotFound:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, f"Task with id {task_id} was not found")
+    return TaskResponse.model_validate(task, from_attributes=True)

tests/api_fastapi/core_api/routes/public/test_dags.py

@@ -274,8 +274,8 @@ class TestDagDetails(TestDagEndpoint):
     @pytest.mark.parametrize(
         "query_params, dag_id, expected_status_code, dag_display_name, start_date",
         [
-            ({}, "fake_dag_id", 404, "fake_dag", datetime(2023, 12, 31, tzinfo=timezone.utc)),
-            ({}, DAG2_ID, 200, DAG2_ID, DAG2_START_DATE),
+            ({}, "fake_dag_id", 404, "fake_dag", "2023-12-31T00:00:00Z"),
+            ({}, DAG2_ID, 200, DAG2_ID, "2021-06-15T00:00:00Z"),
         ],
     )
     def test_dag_details(
@@ -330,7 +330,7 @@ def test_dag_details(
             },
             "render_template_as_native_obj": False,
             "timetable_summary": None,
-            "start_date": start_date.replace(tzinfo=None).isoformat() + "Z",  # pydantic datetime format
+            "start_date": start_date,
             "tags": [],
             "template_search_path": None,
             "timetable_description": "Never, external triggers only",