FastAPI + uvicorn: Is it possible to accept multiple connections with a single worker?

Question

Here is some sample code to demonstrate the issue:

import asyncio
import datetime
import time

import uvicorn
from fastapi import FastAPI
from starlette.responses import PlainTextResponse

app = FastAPI()


@app.get(path="/sync")
def get_sync():
    print(f"sync: {datetime.datetime.now()}: Before sleep")
    time.sleep(5)
    print(f"sync: {datetime.datetime.now()}: After sleep")
    return PlainTextResponse(content=f"sync: {datetime.datetime.now()}: Hello, World!")


@app.get(path="/async")
async def get_async():
    print(f"async: {datetime.datetime.now()}: Before sleep")
    await asyncio.sleep(5)
    print(f"async: {datetime.datetime.now()}: After sleep")
    return PlainTextResponse(content=f"async: {datetime.datetime.now()}: Hello, World!")


if __name__ == "__main__":
    uvicorn.run(app=app, host="0.0.0.0", port=1911)

1. Pick any endpoint above: GET /sync or GET /async
2. Call the endpoint from two different web browser tabs (or use cURL, etc.) to create two parallel requests
3. The first request blocks the second request.

I expected GET /sync to run on a threadpool. I expected GET /async to use some asyncio magic.

I cannot use multiple workers. Is there a solution to allow concurrent requests with a single worker?

FYI: I am using Python 3.7 (64-bit/Win10) and latest versions of FastAPI + unvicorn.

Answer 1

Calling the web endpoint from the browser is going to block (on the client side) since it's waiting for a response. In order to test it properly you need to call it asynchronously. For example you can use a separate python script

import asyncio
import aiohttp
from aiohttp import ClientSession

async def call_async(session: ClientSession, url: str):
    async with session.get(url) as result:

async def main():
    async with aiohttp.ClientSession() as session:
    url = 'https://localhost:5000/async'  # change port 
    tasks = [call_async(session, url) for i in range(5)]
    for finished_task in asyncio.as_completed(tasks):
        print(await finished_task)

asyncio.run(main())