Async & Concurrency

import threading
from typing import Annotated

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient

app = FastAPI()
seen: dict[str, int] = {}


def sync_dep():                        # `def` dependency -> threadpool
    seen["def dependency"] = threading.get_ident()
    return 1


async def async_dep():                 # `async def` dependency -> event loop
    seen["async dependency"] = threading.get_ident()
    return 2


@app.get("/mixed")
def mixed(a: Annotated[int, Depends(sync_dep)], b: Annotated[int, Depends(async_dep)]):
    seen["def path operation"] = threading.get_ident()   # `def` path operation -> threadpool
    return a + b


@app.get("/direct")
async def direct():
    seen["async path operation"] = threading.get_ident()
    return 1


client = TestClient(app)
client.get("/mixed")
client.get("/direct")
loop_thread = seen["async path operation"]
for name, ident in seen.items():
    print(f"{name:22} runs on the event-loop thread: {ident == loop_thread}")

from fastapi import FastAPI

sync_app, async_app = FastAPI(), FastAPI()


@sync_app.get("/books")
def read(limit: int = 20) -> dict:
    return {}


@async_app.get("/books")
async def read(limit: int = 20) -> dict:  # noqa: F811 - same name on purpose
    return {}


print("identical OpenAPI documents:", sync_app.openapi() == async_app.openapi())

import asyncio, time
import httpx
from fastapi import FastAPI

app = FastAPI()


@app.get("/async-sleep")           # cooperative: yields to the event loop while waiting
async def async_sleep():
    await asyncio.sleep(0.2)
    return {"kind": "async def + await asyncio.sleep"}


@app.get("/sync-sleep")            # blocking call, but FastAPI runs `def` in the threadpool
def sync_sleep():
    time.sleep(0.2)
    return {"kind": "def + time.sleep"}


@app.get("/blocking-in-async")     # the bug: blocks the whole event loop
async def blocking_in_async():
    time.sleep(0.2)
    return {"kind": "async def + time.sleep (BAD)"}


async def burst(path: str, n: int = 5) -> float:
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        start = time.perf_counter()
        await asyncio.gather(*(client.get(path) for _ in range(n)))
        return time.perf_counter() - start


async def main():
    for path in ("/async-sleep", "/sync-sleep", "/blocking-in-async"):
        print(f"{path:22} {await burst(path):.2f}s for 5 concurrent requests")

asyncio.run(main())

import asyncio, time
import anyio.to_thread
import httpx
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool

app = FastAPI()


def blocking_report() -> str:       # sync driver, file parsing, requests.get(), ...
    time.sleep(0.2)
    return "done"


@app.get("/threadpool")
async def threadpool():
    result = await run_in_threadpool(blocking_report)       # FastAPI/Starlette helper
    return {"result": result}


@app.get("/anyio")
async def with_anyio():
    result = await anyio.to_thread.run_sync(blocking_report)   # same thing with AnyIO directly
    return {"result": result}


async def main():
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        for path in ("/threadpool", "/anyio"):
            t = time.perf_counter()
            rs = await asyncio.gather(*(client.get(path) for _ in range(5)))
            print(path, [r.json()["result"] for r in rs][:1], f"{time.perf_counter()-t:.2f}s")
    print("default limiter tokens:", anyio.to_thread.current_default_thread_limiter().total_tokens)

asyncio.run(main())

import asyncio
from concurrent.futures import ProcessPoolExecutor

import httpx
from fastapi import FastAPI

pool: ProcessPoolExecutor | None = None


def fib(n: int) -> int:            # CPU-bound: never run this on the event loop
    return n if n < 2 else fib(n - 1) + fib(n - 2)


from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    global pool
    pool = ProcessPoolExecutor(max_workers=2)
    yield
    pool.shutdown()


app = FastAPI(lifespan=lifespan)


@app.get("/fib/{n}")
async def compute(n: int):
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(pool, fib, n)
    return {"n": n, "fib": result}


async def main():
    async with app.router.lifespan_context(app):
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
            print((await c.get("/fib/25")).json())

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import time

import anyio
import httpx
from fastapi import FastAPI

app = FastAPI()


async def fetch_cover(title: str) -> str:
    await anyio.sleep(0.2)               # stands in for an HTTP call to a covers service
    return f"cover-for-{title}"


@app.get("/books/{title}/details")
async def details(title: str):
    results: dict[str, str] = {}

    async def run(name: str, title: str):
        results[name] = await fetch_cover(title)

    async with anyio.create_task_group() as tg:      # structured concurrency (AnyIO)
        tg.start_soon(run, "cover_a", title)
        tg.start_soon(run, "cover_b", title)
        tg.start_soon(run, "cover_c", title)
    return results


async def main():
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
        t = time.perf_counter()
        r = await c.get("/books/dune/details")
        print(r.json(), f"{time.perf_counter() - t:.2f}s")

asyncio.run(main())

import sys
print(sys.version.split()[0], "GIL enabled:", sys._is_gil_enabled())