tl;dr
- Which of the options below is the correct workflow in
fastapi
? - How does one programatically test whether a call is truly blocking (other than manually from browser)? Is there a stress testing extension to
uvicorn
orfastapi
?
I have a number of endpoints in fastapi
server (using uvicorn
at the moment) that have long blocking calls to regular sync Python code. Despite the documentation (https://fastapi.tiangolo.com/async/) I am still unclear whether I should be using exclusively def
, async def
or mixing for my functions.
As far as I understand it, I have three options, assuming:
def some_long_running_sync_function():
...
Option 1 - consistently use def
only for endpoints
@app.get("route/to/endpoint")
def endpoint_1:
some_long_running_sync_function()
@app.post("route/to/another/endpoint")
def endpoint_2:
...
Option 2 - consistently use async def
only and run blocking sync code in executor
import asyncio
@app.get("route/to/endpoint")
async def endpoint_1:
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, some_long_running_sync_function)
@app.post("route/to/another/endpoint")
async def endpoint_2:
...
Option 3 - mix and match def
and async def
based on underlying calls
import asyncio
@app.get("route/to/endpoint")
def endpoint_1:
# endpoint is calling to sync code that cannot be awaited
some_long_running_sync_function()
@app.post("route/to/another/endpoint")
async def endpoint_2:
# this code can be awaited so I can use async
...