Using trio-asyncio with FastAPI

[pquentin]

Two users in chat want to use Trio in FastAPI. FastAPI is quite popular in the async Python world, so I guess they're not the only one.

FastAPI recommends uvicorn for deployment. Unfortunately, uvicorn handles the asyncio event loop itself. How can I use uvicorn, FastAPI and trio-asyncio?

[pquentin]

Writing all the details gave me enough clarity to solve this myself, so I'm answering myself StackOverflow style. The trick is to run uvicorn yourself with the right incantations.

import fastapi
import trio
import trio_asyncio
import uvicorn

app = fastapi.FastAPI()


@app.get("/sleep/{seconds}")
async def sleep(seconds: float):
    await trio_asyncio.trio_as_aio(trio.sleep)(seconds)


if __name__ == "__main__":
    config = uvicorn.Config(app=app)
    server = uvicorn.Server(config=config)
    trio_asyncio.run(trio_asyncio.aio_as_trio(server.serve))

The usual entrypoint is server.run(), but it only setups the asyncio loop and calls the async fucnction server.serve(). So the trick is to simply call server.serve() using trio-asyncio. It's unfortunately not documented, but I guess we could ask uvicorn to document that it's part of the public API.

If you want to deploy that using gunicorn, you can probably subclass uvicorn.workers.UvicornWorker, and change the run() method to use trio_asyncio.run() as above. I haven't tried it, though.

[zuberc-cds]

I tried running this example in windows but it gives following error:

  from trio.hazmat import wait_for_child
{PATH}trio_asyncio\_loop.py:267: TrioAsyncioDeprecationWarning: Using trio-asyncio outside of a Trio event loop is deprecated since trio-asyncio 0.10.0 with no replacement
  return _trio_policy.new_event_loop()
Traceback (most recent call last):
  File ".\server.py", line 20, in <module>
    trio_asyncio.run(trio_asyncio.aio_as_trio(server.serve))
  File "{PATH}trio_asyncio\_loop.py", line 429, in run
    return trio.run(_run_task, proc, args)
  File "{PATH}trio\_core\_run.py", line 1795, in run
    raise runner.main_task_outcome.error
  File "{PATH}trio_asyncio\_loop.py", line 427, in _run_task
    return await proc(*args)
  File "{PATH}trio_asyncio\_adapter.py", line 56, in __call__
    return await self.loop.run_aio_coroutine(f)
  File "{PATH}trio_asyncio\_base.py", line 240, in run_aio_coroutine
    return await run_aio_future(fut)
  File "{PATH}trio_asyncio\_util.py", line 45, in run_aio_future
    res = await trio.hazmat.wait_task_rescheduled(abort_cb)
  File "{PATH}trio\_core\_traps.py", line 165, in wait_task_rescheduled
    return (await _async_yield(WaitTaskRescheduled(abort_func))).unwrap()
  File "{PATH}outcome\_sync.py", line 111, in unwrap
    raise captured_error
  File "{PATH}trio_asyncio\_adapter.py", line 19, in _call_defer
    return await proc(*args, **kwargs)
  File "{PATH}uvicorn\main.py", line 393, in serve
    self.install_signal_handlers()
  File "{PATH}uvicorn\main.py", line 565, in install_signal_handlers
    loop.add_signal_handler(sig, self.handle_exit, sig, None)
  File "{PATH}trio_asyncio\_base.py", line 488, in add_signal_handler
    self._check_signal(sig)
AttributeError: 'TrioEventLoop' object has no attribute '_check_signal'```