You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "omkar-foss (via GitHub)" <gi...@apache.org> on 2024/11/05 19:30:41 UTC
[PR] Migrate public endpoint Get Task to FastAPI [airflow]
omkar-foss opened a new pull request, #43718:
URL: https://github.com/apache/airflow/pull/43718
<!--
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.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
closes: https://github.com/apache/airflow/issues/42874
related: https://github.com/apache/airflow/issues/42370
This migrates the Get Task API from `api_connexion` to `api_fastapi`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1834422605
##########
airflow/api_fastapi/core_api/serializers/tasks.py:
##########
@@ -0,0 +1,159 @@
+# 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
+
+import inspect
+from collections import abc
+from datetime import datetime, timedelta
+
+from pydantic import AliasGenerator, BaseModel, ConfigDict, computed_field, field_validator
+
+from airflow.models.mappedoperator import MappedOperator
+from airflow.serialization.serialized_objects import SerializedBaseOperator, encode_priority_weight_strategy
+from airflow.task.priority_strategy import PriorityWeightStrategy
+
+
+class TimeDeltaResponse(BaseModel):
+ """Time delta serializer for responses."""
+
+ 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),
+ )
+ )
+
+
+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: TimeDeltaResponse | None
+ retry_delay: TimeDeltaResponse | 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("execution_timeout", mode="before")
+ @classmethod
+ def validate_execution_timeout(cls, et: timedelta | None) -> TimeDeltaResponse | None:
+ """Validate the execution_timeout property."""
+ if et is None:
+ return None
+ return TimeDeltaResponse(
+ days=et.days,
+ seconds=et.seconds,
+ microseconds=et.microseconds,
+ )
+
+ @field_validator("retry_delay", mode="before")
+ @classmethod
+ def validate_retry_delay(cls, rd: timedelta | None) -> TimeDeltaResponse | None:
+ """Validate the retry_delay property."""
+ if rd is None:
+ return None
+ return TimeDeltaResponse(
+ days=rd.days,
+ seconds=rd.seconds,
+ microseconds=rd.microseconds,
+ )
+
+ @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()}
+
+ @staticmethod
+ 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,
+ }
+
+ @staticmethod
+ def assign_type_info(task):
+ """Assign type info properties to task and return it."""
+ if task is None:
+ return None
+ task.class_ref = TaskResponse.__get_class_ref(task)
+ task.is_mapped = isinstance(task, MappedOperator)
+ return task
Review Comment:
I've simplified this code, please check it out. Let me know if any changes required, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1832867176
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
FastAPI knows how to run sync code from default async routes. For now all the API code is sync because we are missing some async utility code in airflow core.
This will be changed in the future when we have time to tackle that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on PR #43718:
URL: https://github.com/apache/airflow/pull/43718#issuecomment-2472736918
> Needs rebasing again. Ready to merge
@pierrejeambrun this is done, thank you ✅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1832875205
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
> FastAPI knows how to run sync code from default async routes
That's not true. There is no special magic under the hood of FastAPI. It just wraps all sync endpoints (`def`) into `asyncio.to_thread`, sending them to ThreadPool, and not blocking the current event loop. For `async def` endpoints, it does not do that. If `async def` endpoint calls blocking sync method, the entire event loop will block.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1834546410
##########
airflow/api_fastapi/common/types.py:
##########
@@ -16,10 +16,72 @@
# under the License.
from __future__ import annotations
-from pydantic import AfterValidator, AwareDatetime
+import inspect
+from datetime import timedelta
+
+from pydantic import AfterValidator, AliasGenerator, AwareDatetime, BaseModel, BeforeValidator, ConfigDict
from typing_extensions import Annotated
+from airflow.models.mappedoperator import MappedOperator
+from airflow.models.operator import Operator
+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: Operator) -> 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,
+ }
Review Comment:
Why don't we use field validator as in the legacy code to populate `class_ref` and `operator_name` ? This code should not be in `types`
##########
tests/api_fastapi/core_api/routes/public/test_tasks.py:
##########
@@ -0,0 +1,297 @@
+# 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
+
+import os
+import unittest
+from datetime import datetime
+
+import pytest
+
+from airflow.models.dag import DAG
+from airflow.models.dagbag import DagBag
+from airflow.models.expandinput import EXPAND_INPUT_EMPTY
+from airflow.models.serialized_dag import SerializedDagModel
+from airflow.operators.empty import EmptyOperator
+
+from tests_common.test_utils.db import clear_db_dags, clear_db_runs, clear_db_serialized_dags
+
+pytestmark = pytest.mark.db_test
+
+
+class TestTaskEndpoint:
+ dag_id = "test_dag"
+ mapped_dag_id = "test_mapped_task"
+ unscheduled_dag_id = "test_unscheduled_dag"
+ task_id = "op1"
+ task_id2 = "op2"
+ task_id3 = "op3"
+ mapped_task_id = "mapped_task"
+ unscheduled_task_id1 = "unscheduled_task_1"
+ unscheduled_task_id2 = "unscheduled_task_2"
+ task1_start_date = datetime(2020, 6, 15)
+ task2_start_date = datetime(2020, 6, 16)
+
+ def create_dags(self, test_client):
+ with DAG(self.dag_id, schedule=None, start_date=self.task1_start_date, doc_md="details") as dag:
+ task1 = EmptyOperator(task_id=self.task_id, params={"foo": "bar"})
+ task2 = EmptyOperator(task_id=self.task_id2, start_date=self.task2_start_date)
+
+ with DAG(self.mapped_dag_id, schedule=None, start_date=self.task1_start_date) as mapped_dag:
+ EmptyOperator(task_id=self.task_id3)
+ # Use the private _expand() method to avoid the empty kwargs check.
+ # We don't care about how the operator runs here, only its presence.
+ EmptyOperator.partial(task_id=self.mapped_task_id)._expand(EXPAND_INPUT_EMPTY, strict=False)
+
+ with DAG(self.unscheduled_dag_id, start_date=None, schedule=None) as unscheduled_dag:
+ task4 = EmptyOperator(task_id=self.unscheduled_task_id1, params={"is_unscheduled": True})
+ task5 = EmptyOperator(task_id=self.unscheduled_task_id2, params={"is_unscheduled": True})
+
+ task1 >> task2
+ task4 >> task5
+ dag_bag = DagBag(os.devnull, include_examples=False)
+ dag_bag.dags = {
+ dag.dag_id: dag,
+ mapped_dag.dag_id: mapped_dag,
+ unscheduled_dag.dag_id: unscheduled_dag,
+ }
+ test_client.app.state.dag_bag = dag_bag
+
+ @staticmethod
+ def clear_db():
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
+ @pytest.fixture(autouse=True)
+ def setup(self, test_client) -> None:
+ self.clear_db()
+ self.create_dags(test_client)
+
+ def teardown_method(self) -> None:
+ self.clear_db()
+
+
+class TestGetTask(TestTaskEndpoint):
+ def test_should_respond_200(self, test_client):
+ expected = {
+ "class_ref": {
+ "class_name": "EmptyOperator",
+ "module_path": "airflow.operators.empty",
+ },
+ "depends_on_past": False,
+ "downstream_task_ids": [self.task_id2],
+ "end_date": None,
+ "execution_timeout": None,
+ "extra_links": [],
+ "operator_name": "EmptyOperator",
+ "owner": "airflow",
+ "params": {
+ "foo": {
+ "__class": "airflow.models.param.Param",
+ "value": "bar",
+ "description": None,
+ "schema": {},
+ }
+ },
+ "pool": "default_pool",
+ "pool_slots": 1.0,
+ "priority_weight": 1.0,
+ "queue": "default",
+ "retries": 0.0,
+ "retry_delay": {"__type": "TimeDelta", "days": 0, "seconds": 300, "microseconds": 0},
+ "retry_exponential_backoff": False,
+ "start_date": self.task1_start_date.replace(tzinfo=None).isoformat()
Review Comment:
We can hardcode the expected value. That will be easier than this datetime manipulation to go from isoformat to zulu format which is not great.
Otherwise you can directly convert a `datetime` object to string zulu format with a formatter.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1834608057
##########
airflow/api_fastapi/common/types.py:
##########
@@ -16,10 +16,72 @@
# under the License.
from __future__ import annotations
-from pydantic import AfterValidator, AwareDatetime
+import inspect
+from datetime import timedelta
+
+from pydantic import AfterValidator, AliasGenerator, AwareDatetime, BaseModel, BeforeValidator, ConfigDict
from typing_extensions import Annotated
+from airflow.models.mappedoperator import MappedOperator
+from airflow.models.operator import Operator
+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: Operator) -> 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,
+ }
Review Comment:
Because we don't get the `__class__` and other special properties of `task` in `TaskResponse`, so we need to perform `isinstance()` and other checks before building the `TaskResponse` model.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1831118879
##########
airflow/api_fastapi/core_api/serializers/tasks.py:
##########
@@ -0,0 +1,159 @@
+# 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
+
+import inspect
+from collections import abc
+from datetime import datetime, timedelta
+
+from pydantic import AliasGenerator, BaseModel, ConfigDict, computed_field, field_validator
+
+from airflow.models.mappedoperator import MappedOperator
+from airflow.serialization.serialized_objects import SerializedBaseOperator, encode_priority_weight_strategy
+from airflow.task.priority_strategy import PriorityWeightStrategy
+
+
+class TimeDeltaResponse(BaseModel):
+ """Time delta serializer for responses."""
+
+ 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),
+ )
+ )
Review Comment:
It'll be great if we can remove `__type` from the new API response, since I'm not sure it adds much value from user perspective (since it sort of becomes obvious that it's a timedelta object).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1831803076
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
Calling aync method from async function?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1834621423
##########
airflow/api_fastapi/common/types.py:
##########
@@ -16,10 +16,72 @@
# under the License.
from __future__ import annotations
-from pydantic import AfterValidator, AwareDatetime
+import inspect
+from datetime import timedelta
+
+from pydantic import AfterValidator, AliasGenerator, AwareDatetime, BaseModel, BeforeValidator, ConfigDict
from typing_extensions import Annotated
+from airflow.models.mappedoperator import MappedOperator
+from airflow.models.operator import Operator
+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: Operator) -> 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,
+ }
Review Comment:
@pierrejeambrun where shall I move this code to? Can we keep a `helpers.py` file to have such helper functions or similar?
Since we cannot access the type related properties in the validators, we'll need to call this function before creating `TaskResponse` model.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1833779510
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
> Calling sync method (making network/IO requests) from async function? Doesn't this block the event loop?
Thanks for highlighting this @dolfinus! Yes the sync func will block the event loop for that particular thread. We can probably explore and consider making the sync func async and then call those async ones from the async path funcs. Please see [this comment](https://github.com/apache/airflow/pull/43797#issuecomment-2463876083).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1838610963
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
i think sooner than later we should probably try to add the basic nuts and bolts for asyncio so we can make use of it.
since we're doing all this work migrating code to new thing that supports asyncio, why not just make it async.
here is a "mainly just screwing around" POC for adding some "async" to the scheduler
https://github.com/apache/airflow/pull/43940
so, it seems not too impossible
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1836949954
##########
airflow/api_fastapi/common/types.py:
##########
@@ -16,10 +16,72 @@
# under the License.
from __future__ import annotations
-from pydantic import AfterValidator, AwareDatetime
+import inspect
+from datetime import timedelta
+
+from pydantic import AfterValidator, AliasGenerator, AwareDatetime, BaseModel, BeforeValidator, ConfigDict
from typing_extensions import Annotated
+from airflow.models.mappedoperator import MappedOperator
+from airflow.models.operator import Operator
+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: Operator) -> 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,
+ }
Review Comment:
I did try with a model validator initially, but since the special properties like `__class__` of the `task` object are not available, unfortunately I was unable to get all info required to get `class_ref` initialized.
Please let me know if you've a code snippet that's working to get `__class_` and other properties in the model validator, and I'll try it out. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1834423237
##########
airflow/api_fastapi/core_api/serializers/tasks.py:
##########
@@ -0,0 +1,159 @@
+# 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
+
+import inspect
+from collections import abc
+from datetime import datetime, timedelta
+
+from pydantic import AliasGenerator, BaseModel, ConfigDict, computed_field, field_validator
+
+from airflow.models.mappedoperator import MappedOperator
+from airflow.serialization.serialized_objects import SerializedBaseOperator, encode_priority_weight_strategy
+from airflow.task.priority_strategy import PriorityWeightStrategy
+
+
+class TimeDeltaResponse(BaseModel):
+ """Time delta serializer for responses."""
+
+ 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),
+ )
+ )
+
+
+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: TimeDeltaResponse | None
+ retry_delay: TimeDeltaResponse | 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("execution_timeout", mode="before")
+ @classmethod
+ def validate_execution_timeout(cls, et: timedelta | None) -> TimeDeltaResponse | None:
+ """Validate the execution_timeout property."""
+ if et is None:
+ return None
+ return TimeDeltaResponse(
+ days=et.days,
+ seconds=et.seconds,
+ microseconds=et.microseconds,
+ )
Review Comment:
Done, moved the annotation to `types.py`, using it for both `timedelta` fields.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1834621423
##########
airflow/api_fastapi/common/types.py:
##########
@@ -16,10 +16,72 @@
# under the License.
from __future__ import annotations
-from pydantic import AfterValidator, AwareDatetime
+import inspect
+from datetime import timedelta
+
+from pydantic import AfterValidator, AliasGenerator, AwareDatetime, BaseModel, BeforeValidator, ConfigDict
from typing_extensions import Annotated
+from airflow.models.mappedoperator import MappedOperator
+from airflow.models.operator import Operator
+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: Operator) -> 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,
+ }
Review Comment:
@pierrejeambrun where shall I move this code to? Can we keep a helpers file to have such helper functions or so.
Since we cannot access the type related properties in the validators, we'll need to call this function before creating `TaskResponse` model.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1833779510
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
> Calling sync method (making network/IO requests) from async function? Doesn't this block the event loop?
Thanks for highlighting this @dolfinus! Yes the sync func will block the event loop for that particular thread. We can probably explore and consider wrapping the sync funcs in async and then call those from the async path funcs. Please see [this comment](https://github.com/apache/airflow/pull/43797#issuecomment-2463876083).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1831803076
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
Calling sync method (making network/IO requests) from async function? Doesn't this block the event loop?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1830614915
##########
airflow/api_fastapi/core_api/serializers/tasks.py:
##########
@@ -0,0 +1,159 @@
+# 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
+
+import inspect
+from collections import abc
+from datetime import datetime, timedelta
+
+from pydantic import AliasGenerator, BaseModel, ConfigDict, computed_field, field_validator
+
+from airflow.models.mappedoperator import MappedOperator
+from airflow.serialization.serialized_objects import SerializedBaseOperator, encode_priority_weight_strategy
+from airflow.task.priority_strategy import PriorityWeightStrategy
+
+
+class TimeDeltaResponse(BaseModel):
+ """Time delta serializer for responses."""
+
+ 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),
+ )
+ )
+
+
+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: TimeDeltaResponse | None
+ retry_delay: TimeDeltaResponse | 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("execution_timeout", mode="before")
+ @classmethod
+ def validate_execution_timeout(cls, et: timedelta | None) -> TimeDeltaResponse | None:
+ """Validate the execution_timeout property."""
+ if et is None:
+ return None
+ return TimeDeltaResponse(
+ days=et.days,
+ seconds=et.seconds,
+ microseconds=et.microseconds,
+ )
Review Comment:
This can be factorized with `Annotated[...., BeforeValidator]` notation. (same for retry_delay).
1 common function, used for 3 different fields.
##########
airflow/api_fastapi/core_api/serializers/tasks.py:
##########
@@ -0,0 +1,159 @@
+# 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
+
+import inspect
+from collections import abc
+from datetime import datetime, timedelta
+
+from pydantic import AliasGenerator, BaseModel, ConfigDict, computed_field, field_validator
+
+from airflow.models.mappedoperator import MappedOperator
+from airflow.serialization.serialized_objects import SerializedBaseOperator, encode_priority_weight_strategy
+from airflow.task.priority_strategy import PriorityWeightStrategy
+
+
+class TimeDeltaResponse(BaseModel):
+ """Time delta serializer for responses."""
+
+ 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),
+ )
+ )
Review Comment:
Why do we need this ?
##########
airflow/api_fastapi/core_api/serializers/tasks.py:
##########
@@ -0,0 +1,159 @@
+# 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
+
+import inspect
+from collections import abc
+from datetime import datetime, timedelta
+
+from pydantic import AliasGenerator, BaseModel, ConfigDict, computed_field, field_validator
+
+from airflow.models.mappedoperator import MappedOperator
+from airflow.serialization.serialized_objects import SerializedBaseOperator, encode_priority_weight_strategy
+from airflow.task.priority_strategy import PriorityWeightStrategy
+
+
+class TimeDeltaResponse(BaseModel):
+ """Time delta serializer for responses."""
+
+ 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),
+ )
+ )
+
+
+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: TimeDeltaResponse | None
+ retry_delay: TimeDeltaResponse | 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("execution_timeout", mode="before")
+ @classmethod
+ def validate_execution_timeout(cls, et: timedelta | None) -> TimeDeltaResponse | None:
+ """Validate the execution_timeout property."""
+ if et is None:
+ return None
+ return TimeDeltaResponse(
+ days=et.days,
+ seconds=et.seconds,
+ microseconds=et.microseconds,
+ )
+
+ @field_validator("retry_delay", mode="before")
+ @classmethod
+ def validate_retry_delay(cls, rd: timedelta | None) -> TimeDeltaResponse | None:
+ """Validate the retry_delay property."""
+ if rd is None:
+ return None
+ return TimeDeltaResponse(
+ days=rd.days,
+ seconds=rd.seconds,
+ microseconds=rd.microseconds,
+ )
+
+ @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()}
+
+ @staticmethod
+ 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,
+ }
+
+ @staticmethod
+ def assign_type_info(task):
+ """Assign type info properties to task and return it."""
+ if task is None:
+ return None
+ task.class_ref = TaskResponse.__get_class_ref(task)
+ task.is_mapped = isinstance(task, MappedOperator)
+ return task
Review Comment:
You can re-use the `_call_function` validator (put it in a common code place and update signature with generic types (typevars)). This way you can do.
`Annotated[dict, BeforeValidator(_call_function)]`
Anyway we shouldn't have a `staticmethod` to call to populate the field. Lets use validators as we do for other fields. (field or model validator)
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
Review Comment:
I think it's fine to do it that way, this is how we do it for other routers.
I think we can even add to the prefix the `/{dag_id}`, all tasks endpoint need it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on PR #43718:
URL: https://github.com/apache/airflow/pull/43718#issuecomment-2465200380
> Overall looking good. Need rebasing and conflicts resolution. Thanks
Done! Rebased and conflicts resolved ✅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1834423237
##########
airflow/api_fastapi/core_api/serializers/tasks.py:
##########
@@ -0,0 +1,159 @@
+# 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
+
+import inspect
+from collections import abc
+from datetime import datetime, timedelta
+
+from pydantic import AliasGenerator, BaseModel, ConfigDict, computed_field, field_validator
+
+from airflow.models.mappedoperator import MappedOperator
+from airflow.serialization.serialized_objects import SerializedBaseOperator, encode_priority_weight_strategy
+from airflow.task.priority_strategy import PriorityWeightStrategy
+
+
+class TimeDeltaResponse(BaseModel):
+ """Time delta serializer for responses."""
+
+ 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),
+ )
+ )
+
+
+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: TimeDeltaResponse | None
+ retry_delay: TimeDeltaResponse | 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("execution_timeout", mode="before")
+ @classmethod
+ def validate_execution_timeout(cls, et: timedelta | None) -> TimeDeltaResponse | None:
+ """Validate the execution_timeout property."""
+ if et is None:
+ return None
+ return TimeDeltaResponse(
+ days=et.days,
+ seconds=et.seconds,
+ microseconds=et.microseconds,
+ )
Review Comment:
Done, added common function to `types.py`, using it for both `timedelta` fields.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1835330707
##########
airflow/api_fastapi/common/types.py:
##########
@@ -16,10 +16,72 @@
# under the License.
from __future__ import annotations
-from pydantic import AfterValidator, AwareDatetime
+import inspect
+from datetime import timedelta
+
+from pydantic import AfterValidator, AliasGenerator, AwareDatetime, BaseModel, BeforeValidator, ConfigDict
from typing_extensions import Annotated
+from airflow.models.mappedoperator import MappedOperator
+from airflow.models.operator import Operator
+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: Operator) -> 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,
+ }
Review Comment:
On a model validator you should be able to access the task I think.
This will push this code into the serialised and avoid having this on its own somewhere that will most likely not be reused.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #43718:
URL: https://github.com/apache/airflow/pull/43718#issuecomment-2464969264
Branch needs rebasing and conflict solving.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1834608057
##########
airflow/api_fastapi/common/types.py:
##########
@@ -16,10 +16,72 @@
# under the License.
from __future__ import annotations
-from pydantic import AfterValidator, AwareDatetime
+import inspect
+from datetime import timedelta
+
+from pydantic import AfterValidator, AliasGenerator, AwareDatetime, BaseModel, BeforeValidator, ConfigDict
from typing_extensions import Annotated
+from airflow.models.mappedoperator import MappedOperator
+from airflow.models.operator import Operator
+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: Operator) -> 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,
+ }
Review Comment:
Because we don't get the `__class__` and other properties of `task` in `TaskResponse`, so we need to perform `isinstance` and other checks before building the `TaskResponse` model.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1832875205
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
> FastAPI knows how to run sync code from default async routes
That's not true. There is no special magic under the hood of FastAPI. It just wraps all sync endpoints (`def`) into `asyncio.to_thread`, sending them to ThreadPool, and not blocking the current event loop. For `async def` endpoints, it does not do that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1836955834
##########
airflow/api_fastapi/common/types.py:
##########
@@ -16,10 +16,72 @@
# under the License.
from __future__ import annotations
-from pydantic import AfterValidator, AwareDatetime
+import inspect
+from datetime import timedelta
+
+from pydantic import AfterValidator, AliasGenerator, AwareDatetime, BaseModel, BeforeValidator, ConfigDict
from typing_extensions import Annotated
+from airflow.models.mappedoperator import MappedOperator
+from airflow.models.operator import Operator
+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: Operator) -> 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,
+ }
Review Comment:
Alternatively, it'll be great if can take this up with next PR (Get Tasks) and wrap up this one! (Need models from this PR in Get Tasks 🙂)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1829939833
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
Review Comment:
Hi, please let me know if using `/dags` prefix makes sense here, or if `tasks_router` should be added as a child to `dags_router` (or if there's another better way to do this).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1829939833
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
Review Comment:
Hi, let me know if using `/dags` prefix makes sense here, or if `tasks_router` should be added as a child to dags_router (or if there's another better way to do this).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1833779510
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
> Calling sync method (making network/IO requests) from async function? Doesn't this block the event loop?
Thanks for highlighting this @dolfinus! Yes the sync func will block the event loop for that particular thread. We can probably explore and consider making the sync funcs async, and then can call those async ones from the async path funcs. Please see [this comment](https://github.com/apache/airflow/pull/43797#issuecomment-2463876083).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1834619178
##########
tests/api_fastapi/core_api/routes/public/test_tasks.py:
##########
@@ -0,0 +1,297 @@
+# 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
+
+import os
+import unittest
+from datetime import datetime
+
+import pytest
+
+from airflow.models.dag import DAG
+from airflow.models.dagbag import DagBag
+from airflow.models.expandinput import EXPAND_INPUT_EMPTY
+from airflow.models.serialized_dag import SerializedDagModel
+from airflow.operators.empty import EmptyOperator
+
+from tests_common.test_utils.db import clear_db_dags, clear_db_runs, clear_db_serialized_dags
+
+pytestmark = pytest.mark.db_test
+
+
+class TestTaskEndpoint:
+ dag_id = "test_dag"
+ mapped_dag_id = "test_mapped_task"
+ unscheduled_dag_id = "test_unscheduled_dag"
+ task_id = "op1"
+ task_id2 = "op2"
+ task_id3 = "op3"
+ mapped_task_id = "mapped_task"
+ unscheduled_task_id1 = "unscheduled_task_1"
+ unscheduled_task_id2 = "unscheduled_task_2"
+ task1_start_date = datetime(2020, 6, 15)
+ task2_start_date = datetime(2020, 6, 16)
+
+ def create_dags(self, test_client):
+ with DAG(self.dag_id, schedule=None, start_date=self.task1_start_date, doc_md="details") as dag:
+ task1 = EmptyOperator(task_id=self.task_id, params={"foo": "bar"})
+ task2 = EmptyOperator(task_id=self.task_id2, start_date=self.task2_start_date)
+
+ with DAG(self.mapped_dag_id, schedule=None, start_date=self.task1_start_date) as mapped_dag:
+ EmptyOperator(task_id=self.task_id3)
+ # Use the private _expand() method to avoid the empty kwargs check.
+ # We don't care about how the operator runs here, only its presence.
+ EmptyOperator.partial(task_id=self.mapped_task_id)._expand(EXPAND_INPUT_EMPTY, strict=False)
+
+ with DAG(self.unscheduled_dag_id, start_date=None, schedule=None) as unscheduled_dag:
+ task4 = EmptyOperator(task_id=self.unscheduled_task_id1, params={"is_unscheduled": True})
+ task5 = EmptyOperator(task_id=self.unscheduled_task_id2, params={"is_unscheduled": True})
+
+ task1 >> task2
+ task4 >> task5
+ dag_bag = DagBag(os.devnull, include_examples=False)
+ dag_bag.dags = {
+ dag.dag_id: dag,
+ mapped_dag.dag_id: mapped_dag,
+ unscheduled_dag.dag_id: unscheduled_dag,
+ }
+ test_client.app.state.dag_bag = dag_bag
+
+ @staticmethod
+ def clear_db():
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
+ @pytest.fixture(autouse=True)
+ def setup(self, test_client) -> None:
+ self.clear_db()
+ self.create_dags(test_client)
+
+ def teardown_method(self) -> None:
+ self.clear_db()
+
+
+class TestGetTask(TestTaskEndpoint):
+ def test_should_respond_200(self, test_client):
+ expected = {
+ "class_ref": {
+ "class_name": "EmptyOperator",
+ "module_path": "airflow.operators.empty",
+ },
+ "depends_on_past": False,
+ "downstream_task_ids": [self.task_id2],
+ "end_date": None,
+ "execution_timeout": None,
+ "extra_links": [],
+ "operator_name": "EmptyOperator",
+ "owner": "airflow",
+ "params": {
+ "foo": {
+ "__class": "airflow.models.param.Param",
+ "value": "bar",
+ "description": None,
+ "schema": {},
+ }
+ },
+ "pool": "default_pool",
+ "pool_slots": 1.0,
+ "priority_weight": 1.0,
+ "queue": "default",
+ "retries": 0.0,
+ "retry_delay": {"__type": "TimeDelta", "days": 0, "seconds": 300, "microseconds": 0},
+ "retry_exponential_backoff": False,
+ "start_date": self.task1_start_date.replace(tzinfo=None).isoformat()
Review Comment:
Done ✅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1834422007
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
Review Comment:
Done, using prefix `/dags/{dag_id}/tasks` since it's common for both Get Task and Get Tasks endpoints.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1832957241
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
True, I got confused, I thought It was run in an external threadpool, but that's only true for `non async` route.I think we need to remove all of the `async` keyword as long as the code has blocking sync operations.
I'll open a PR for that, thanks for pointing it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on PR #43718:
URL: https://github.com/apache/airflow/pull/43718#issuecomment-2464819740
@pierrejeambrun PR rebased and synced with `main` and all PR comments resolved, please check it out. Thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "omkar-foss (via GitHub)" <gi...@apache.org>.
omkar-foss commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1830983668
##########
airflow/api_fastapi/core_api/serializers/tasks.py:
##########
@@ -0,0 +1,159 @@
+# 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
+
+import inspect
+from collections import abc
+from datetime import datetime, timedelta
+
+from pydantic import AliasGenerator, BaseModel, ConfigDict, computed_field, field_validator
+
+from airflow.models.mappedoperator import MappedOperator
+from airflow.serialization.serialized_objects import SerializedBaseOperator, encode_priority_weight_strategy
+from airflow.task.priority_strategy import PriorityWeightStrategy
+
+
+class TimeDeltaResponse(BaseModel):
+ """Time delta serializer for responses."""
+
+ 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),
+ )
+ )
Review Comment:
Currently, the `TimeDelta` schema has an `objectType` property which gets serialized as `__type`, code [here](https://github.com/apache/airflow/blob/main/airflow/api_connexion/schemas/common_schema.py#L41). So, we need this to maintain consistency with old API's `TimeDelta` schema, example in test [here](https://github.com/apache/airflow/blob/main/tests/api_connexion/endpoints/test_task_endpoint.py#L135).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1831803076
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
Calling aync method from async function? Doesn't this block the event loop?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun merged PR #43718:
URL: https://github.com/apache/airflow/pull/43718
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1838904729
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
ok @pierrejeambrun i have added an asyncio "hello world" for airflow
maybe then we can actually implement async as we migrate????
https://github.com/apache/airflow/pull/43944
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Migrate public endpoint Get Task to FastAPI [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43718:
URL: https://github.com/apache/airflow/pull/43718#discussion_r1833120287
##########
airflow/api_fastapi/core_api/routes/public/tasks.py:
##########
@@ -0,0 +1,54 @@
+# 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.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.serializers.tasks import TaskResponse
+from airflow.exceptions import TaskNotFound
+from airflow.models import DAG
+
+tasks_router = AirflowRouter(tags=["Task"], prefix="/dags")
+
+
+@tasks_router.get(
+ "/{dag_id}/tasks/{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,
+ ]
+ ),
+)
+async 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)
Review Comment:
PR is ready here, https://github.com/apache/airflow/pull/43797
Thanks again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org