Fastapi 0.123.5 fails tests

[potiuk]

Body

FastAPI 0.123.5 breaks system tests with:

  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/errors.py", line 164, in __call__
    |     await self.app(scope, receive, _send)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 191, in __call__
    |     with recv_stream, send_stream, collapse_excgroups():
    |   File "/usr/python/lib/python3.10/contextlib.py", line 153, in __exit__
    |     self.gen.throw(typ, value, traceback)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/_utils.py", line 85, in collapse_excgroups
    |     raise exc
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 193, in __call__
    |     response = await self.dispatch_func(request, call_next)
    |   File "/opt/airflow/airflow-core/src/airflow/api_fastapi/execution_api/app.py", line 134, in dispatch
    |     response: Response = await call_next(request)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 168, in call_next
    |     raise app_exc from app_exc.__cause__ or app_exc.__context__
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 144, in coro
    |     await self.app(scope, receive_or_disconnect, send_no_error)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 191, in __call__
    |     with recv_stream, send_stream, collapse_excgroups():
    |   File "/usr/python/lib/python3.10/contextlib.py", line 153, in __exit__
    |     self.gen.throw(typ, value, traceback)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/_utils.py", line 85, in collapse_excgroups
    |     raise exc
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 193, in __call__
    |     response = await self.dispatch_func(request, call_next)
    |   File "/opt/airflow/airflow-core/src/airflow/api_fastapi/execution_api/app.py", line 122, in dispatch
    |     response: Response = await call_next(request)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 168, in call_next
    |     raise app_exc from app_exc.__cause__ or app_exc.__context__
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 144, in coro
    |     await self.app(scope, receive_or_disconnect, send_no_error)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 191, in __call__
    |     with recv_stream, send_stream, collapse_excgroups():
    |   File "/usr/python/lib/python3.10/contextlib.py", line 153, in __exit__
    |     self.gen.throw(typ, value, traceback)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/_utils.py", line 85, in collapse_excgroups
    |     raise exc
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 193, in __call__
    |     response = await self.dispatch_func(request, call_next)
    |   File "/usr/python/lib/python3.10/site-packages/cadwyn/middleware.py", line 124, in dispatch
    |     response = await call_next(request)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 168, in call_next
    |     raise app_exc from app_exc.__cause__ or app_exc.__context__
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/base.py", line 144, in coro
    |     await self.app(scope, receive_or_disconnect, send_no_error)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/middleware/exceptions.py", line 63, in __call__
    |     await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    |     raise exc
    |   File "/usr/python/lib/python3.10/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
    |     await app(scope, receive, sender)
    |   File "/usr/python/lib/python3.10/site-packages/fastapi/middleware/asyncexitstack.py", line 18, in __call__
    |     await self.app(scope, receive, send)
    |   File "/usr/python/lib/python3.10/site-packages/cadwyn/routing.py", line 93, in __call__
    |     await self.process_request(scope=scope, receive=receive, send=send, routes=routes)
    |   File "/usr/python/lib/python3.10/site-packages/cadwyn/routing.py", line 131, in process_request
    |     await route.handle(scope, receive, send)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/routing.py", line 290, in handle
    |     await self.app(scope, receive, send)
    |   File "/usr/python/lib/python3.10/site-packages/fastapi/routing.py", line 119, in app
    |     await wrap_app_handling_exceptions(app, request)(scope, receive, send)
    |   File "/usr/python/lib/python3.10/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
    |     raise exc
    |   File "/usr/python/lib/python3.10/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
    |     await app(scope, receive, sender)
    |   File "/usr/python/lib/python3.10/site-packages/fastapi/routing.py", line 105, in app
    |     response = await f(request)
    |   File "/usr/python/lib/python3.10/site-packages/fastapi/routing.py", line 407, in app
    |     content = await serialize_response(
    |   File "/usr/python/lib/python3.10/site-packages/fastapi/routing.py", line 248, in serialize_response
    |     raise ResponseValidationError(
    | fastapi.exceptions.ResponseValidationError: 1 validation errors:
    |   {'type': 'model_attributes_type', 'loc': ('response',), 'msg': 'Input should be a valid dictionary or object to extract fields from', 'input': <coroutine object ti_run at 0x7f35577a15b0>}

Example here:

https://github.com/apache/airflow/actions/runs/19875061754/job/56961409679?pr=58976

Likely caused by this change: fastapi/fastapi#14434

Comment in the issue - needs further investigation, in the meantime we added != on the version in #58980

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[potiuk]

cc: @pierrejeambrun, this likely need some deepar analysis. I commented on the PR in fastapi, but it's hard to see what could be wrong.

[tirkarthi]

This seems to be related to fastapi/fastapi#14434 . The wrapped call is validated to be a coroutine. Adding print statements in the PR yields below results where there is a difference in behavior related to the function being a coroutine or not. After the fastapi PR the __wrapped__ part is validated for coroutine check.

iscoroutinefunction(dependant.call) = True
iscoroutinefunction(dependant._unwrapped_call) = False
dependant.is_coroutine_callable = False
dependant._unwrapped_call = <function ti_run at 0x7f3e1fe7f880>
dependant.call = <function ti_run at 0x7f3e1df48220>
inspect.isroutine(dependant._unwrapped_call) = True
dir(dependant.call) = ['__annotations__', '__builtins__', '__call__', '__class__', '__closure__', '__code__', '__defaults__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__get__', '__getattribute__', '__getstate__', '__globals__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__kwdefaults__', '__le__', '__lt__', '__module__', '__name__', '__ne__', '__new__', '__qualname__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__signature__', '__sizeof__', '__str__', '__subclasshook__', '__wrapped__', '_original_callable']
dir(dependant._unwrapped_call) = ['__annotations__', '__builtins__', '__call__', '__class__', '__closure__', '__code__', '__defaults__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__get__', '__getattribute__', '__getstate__', '__globals__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__kwdefaults__', '__le__', '__lt__', '__module__', '__name__', '__ne__', '__new__', '__qualname__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__']
dependant.call.__wrapped__ = <cadwyn.schema_generation._CallableWrapper object at 0x7f3e1ef59350>

[tirkarthi]

Looks like an issue with cadwyn and the non coroutine methods being wrapped as _CallableWrapper . Applying the following changes of defining the functions as async _AsyncCallableWrapper is set and tests are passing for airflow-core/tests/unit/models/test_mappedoperator.py::test_one_to_many_as_teardown

diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index 56111db979..0cc2e45b3a 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -101,7 +101,7 @@ log = structlog.get_logger(__name__)
     },
     response_model_exclude_unset=True,
 )
-def ti_run(
+async def ti_run(
     task_instance_id: UUID,
     ti_run_payload: Annotated[TIEnterRunningPayload, Body()],
     session: SessionDep,
@@ -341,7 +341,7 @@ def _get_upstream_map_indexes(
         HTTP_422_UNPROCESSABLE_CONTENT: {"description": "Invalid payload for the state transition"},
     },
 )
-def ti_update_state(
+async def ti_update_state(
     task_instance_id: UUID,
     ti_patch_payload: Annotated[TIStateUpdate, Body()],
     session: SessionDep,
@@ -709,7 +709,7 @@ def ti_heartbeat(
         },
     },
 )
-def ti_put_rtif(
+async def ti_put_rtif(
     task_instance_id: UUID,
     put_rtif_payload: Annotated[dict[str, JsonValue], Body()],
     session: SessionDep,
@@ -739,7 +739,7 @@ def ti_put_rtif(
         HTTP_422_UNPROCESSABLE_CONTENT: {"description": "Invalid rendered_map_index value"},
     },
 )
-def ti_patch_rendered_map_index(
+async def ti_patch_rendered_map_index(
     task_instance_id: UUID,
     rendered_map_index: Annotated[str, Body()],
     session: SessionDep,
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py b/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
index ce8309341f..d31c7c7728 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py
@@ -134,7 +134,7 @@ class GetXcomFilterParams(BaseModel):
     "/{dag_id}/{run_id}/{task_id}/{key}",
     description="Get a single XCom Value",
 )
-def get_xcom(
+async def get_xcom(
     dag_id: str,
     run_id: str,
     task_id: str,
@@ -316,7 +316,7 @@ def get_mapped_xcom_by_slice(
     "/{dag_id}/{run_id}/{task_id}/{key}",
     status_code=status.HTTP_201_CREATED,
 )
-def set_xcom(
+async def set_xcom(
     dag_id: str,
     run_id: str,
     task_id: str,

[potiuk]

Nice investigation. Would love to learnn more :)

Doeas it mean that we should define those method as async in general ? Or is it something that cadwyn should fix ? I understand that cadwyn auto-wrapping out API simply changes our sync methods into async ones from the fastapi point of view ?

[tirkarthi]

I have opened an issue in cadwyn repo zmievsa/cadwyn#309 with below reproducer from one of the examples using sync. The below works in fastapi==0.123.4 but breaks in fastapi==0.123.5 using sync.

import uuid

from pydantic import BaseModel

from cadwyn import (
    Cadwyn,
    Version,
    VersionBundle,
    VersionedAPIRouter,
)


class UserCreateRequest(BaseModel):
    address: str


class UserResource(BaseModel):
    id: uuid.UUID
    address: str


database_parody = {}
router = VersionedAPIRouter()


@router.post("/users")
def create_user(payload: UserCreateRequest) -> UserResource:
    id_ = uuid.uuid4()
    database_parody[id_] = UserResource(id=id_, address=payload.address)
    return database_parody[id_]


@router.get("/users/{user_id}")
def get_user(user_id: uuid.UUID) -> UserResource:
    return database_parody[user_id]


app = Cadwyn(versions=VersionBundle(Version("2000-01-01")))
app.generate_and_include_versioned_routers(router)

[potiuk]

Yeah. I let the Fast API team know - interesting case :)

[potiuk]

Nice, small reproducer. Good one @tirkarthi !

[potiuk]

database_parody

🤣

[pierrejeambrun]

Thanks @poituk and @tirkarthi for opening the relevant issues

[bugraoz93]

Thanks a lot @potiuk and @tirkarthi for following up with the issues!