[QUESTION]:used AsyncMilvusClient in FastAPI

[jitengfei]

Is there an existing issue for this?

• I have searched the existing issues

What is your question?

milvus version 2.4.12
pymilvus version 2.5.4

When I used AsyncMilvusClient in FastAPI, I encountered the following error:
ERROR: Exception in ASGI application
Traceback (most recent call last):
File "d:\projects\test\milvus_test\env\lib\site-packages\uvicorn\protocols\http\h11_impl.py", line 406, in run_asgi
result = await app( # type: ignore[func-returns-value]
File "d:\projects\test\milvus_test\env\lib\site-packages\uvicorn\middleware\proxy_headers.py", line 60, in call
return await self.app(scope, receive, send)
File "d:\projects\test\milvus_test\env\lib\site-packages\fastapi\applications.py", line 1054, in call
await super().call(scope, receive, send)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette\applications.py", line 113, in call
await self.middleware_stack(scope, receive, send)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette\middleware\errors.py", line 187, in call
raise exc
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette\middleware\errors.py", line 165, in call
await self.app(scope, receive, _send)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette\middleware\gzip.py", line 20, in call
await responder(scope, receive, send)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette\middleware\gzip.py", line 39, in call
await self.app(scope, receive, self.send_with_gzip)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette\middleware\exceptions.py", line 62, in call
await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette_exception_handler.py", line 62, in wrapped_app
raise exc
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette_exception_handler.py", line 51, in wrapped_app
await app(scope, receive, sender)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette\routing.py", line 715, in call
await self.middleware_stack(scope, receive, send)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette\routing.py", line 735, in app
await route.handle(scope, receive, send)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette\routing.py", line 288, in handle
await self.app(scope, receive, send)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette\routing.py", line 76, in app
await wrap_app_handling_exceptions(app, request)(scope, receive, send)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette_exception_handler.py", line 62, in wrapped_app
raise exc
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette_exception_handler.py", line 51, in wrapped_app
await app(scope, receive, sender)
File "d:\projects\test\milvus_test\env\lib\site-packages\starlette\routing.py", line 73, in app
response = await f(request)
File "d:\projects\test\milvus_test\env\lib\site-packages\fastapi\routing.py", line 301, in app
raw_response = await run_endpoint_function(
File "d:\projects\test\milvus_test\env\lib\site-packages\fastapi\routing.py", line 212, in run_endpoint_function
return await dependant.call(**values)
File "D:\projects\test\milvus_test\controller\test.py", line 16, in search
result = await get_bing_web_search_cache(q="")
File "D:\projects\test\milvus_test\core\milvus_test.py", line 22, in get_bing_web_search_cache
res = await client.search(
File "d:\projects\test\milvus_test\env\lib\site-packages\pymilvus\milvus_client\async_milvus_client.py", line 394, in search
raise ex from ex
File "d:\projects\test\milvus_test\env\lib\site-packages\pymilvus\milvus_client\async_milvus_client.py", line 379, in search
res = await conn.search(
File "d:\projects\test\milvus_test\env\lib\site-packages\pymilvus\client\async_grpc_handler.py", line 600, in search
return await self._execute_search(request, timeout, round_decimal=round_decimal, **kwargs)
File "d:\projects\test\milvus_test\env\lib\site-packages\pymilvus\client\async_grpc_handler.py", line 548, in _execute_search
raise e from e
File "d:\projects\test\milvus_test\env\lib\site-packages\pymilvus\client\async_grpc_handler.py", line 538, in _execute_search
response = await self._async_stub.Search(request, timeout=timeout)
File "d:\projects\test\milvus_test\env\lib\site-packages\grpc\aio_interceptor.py", line 471, in await
call = yield from self._interceptors_task.await()
RuntimeError: Task <Task pending name='Task-12' coro=<RequestResponseCycle.run_asgi() running at d:\projects\test\milvus_test\env\lib\site-packages\uvicorn\protocols\http\h11_impl.py:406> cb=[set.discard()]> got Future <Task pending name='Task-13' coro=<InterceptedUnaryUnaryCall._invoke() running at d:\projects\test\milvus_test\env\lib\site-packages\grpc\aio_interceptor.py:652> cb=[InterceptedCall._fire_or_add_pending_done_callbacks()]> attached to a different loop

Anything else?

No response

[jitengfei]

fastapi==0.115.0
uvicorn==0.31.0
pymilvus[midel]==2.5.4

[XuanYang-cn]

/assign @brcarry
PTAL

[sre-ci-robot]

@XuanYang-cn: GitHub didn't allow me to assign the following users: brcarry.

Note that only milvus-io members, repo collaborators and people who have commented on this issue/PR can be assigned. Additionally, issues/PRs can only have 10 assignees at the same time.
For more information please see the contributor guide

In response to this:

/assign @brcarry
PTAL

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.

[brcarry]

@jitengfei Could you show me how you use the AsyncMilvusClient in your code?

[liubaiand]

通过这个调fastapi CMD ["gunicorn", "main:app", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:9088","--workers", "4","--threads","2","--timeout","2000","--capture-output"]
代码中正常使用 AsyncMilvusClient
@brcarry

[liubaiand]

@jitengfei May I ask if the issue has been resolved?

[ahmadcolourbox]

A workaround is to use send it to to_thread:

async def async_search(self, **kwargs):
    def run_sync_search():
        self.client.load_collection(kwargs.get('collection_name'))
        return self.client.search(**kwargs)

    result = await asyncio.to_thread(run_sync_search)

    return result

[iamcodingcat]

I encountered this error with async api in fastapi + uvicorn + gunicorn.(pymilvus version 2.5.8)
@ahmadcolourbox I ask for your code snippet. your function self.client.search is sync api not async, right?

[ahmadcolourbox]

@iamcodingcat Yes, just use the sync library not the async.

A better approach is to use decorators:

def make_async(func: Callable) -> Any:
    @functools.wraps(func)
    async def async_wrapper(*args, **kwargs) -> Any:
        return await asyncio.to_thread(functools.partial(func, *args, **kwargs))
    return async_wrapper

And then in your client file that connect to Milvus create a search function and wrap it up with the make_async decorator:

@make_async
def async_search(self, **kwargs):
    return self.client.search(**kwargs)

From FastAPI:

results = await async_search(...)