The path to AI FDE #04: Many calls at once, and a FastAPI service you can lean on

Posted on Oct 22 • Originally published at dhanushkumarsivaji.substack.com

Issue #03 ended with tasks: the way async work actually overlaps. Real AI backends fan out, calling a model several times or fetching a few sources in parallel. So this issue starts with how to do that without losing errors or hanging forever. Then it turns to the FastAPI features that move an endpoint past hello-world. As always, every example is a toy one (a notes app), not my roadmap exercises.

1. gather returns results in the order you asked. a, b = await asyncio.gather(fetch(1), fetch(2)) runs both concurrently, and the result list matches the order of the arguments, not the order the calls finished in. It’s the closest thing to Promise.all.

2. Know what gather does with an error. By default, the first exception is raised straight to you, and the other calls are not cancelled: they keep running in the background. With return_exceptions=True, errors come back as items in the result list instead, which suits best-effort work.

3. TaskGroup is all-or-nothing. Inside async with asyncio.TaskGroup() as tg:, you start work with tg.create_task. The first failure cancels the rest, and when the block exits the failures are raised together as an ExceptionGroup, which you handle with except* SomeError:. Use it when partial results are useless. It also keeps task references for you (#03, point 11).

4. Every network wait needs a deadline. async with asyncio.timeout(10): cancels whatever is inside after 10 seconds and raises TimeoutError, which you can only catch outside the block. For a single awaitable, await asyncio.wait_for(coro, timeout=10) does the same job. Without a deadline, one slow provider call can hold a request open indefinitely.

5. A semaphore caps how many run at once. sem = asyncio.Semaphore(5), then async with sem: around each call, lets at most five run concurrently while the rest wait their turn. That’s how you fan out over 200 items while keeping concurrency low enough to stay under a provider’s rate limit (the 429s from #01).

6. Don’t swallow cancellation. TaskGroup and timeout() work by cancelling tasks, and the asyncio docs warn that they can misbehave if your code catches and hides asyncio.CancelledError. A bare except:, an except BaseException: or a re-raise you forgot quietly breaks your deadlines. ( except Exception: doesn’t catch it: since Python 3.8, CancelledError is a BaseException.) Catch specific errors (#02, point 4).

7. Request models give you 422s for free. Declare a body parameter as a Pydantic model ( note: NoteIn ), and FastAPI validates the request, answering 422 with a list of what’s wrong before your function even runs.

8. Response models validate and filter what goes out. Set a return type or response_model=NoteOut, and FastAPI checks your output and drops any field the model doesn’t declare. That’s the docs’ security point: an internal field (an owner ID, a password hash) can’t leak just because it was on the object you returned. If you set both, response_model takes priority.

9. HTTPException for errors you mean. raise HTTPException(status_code=404, detail="Note not found") stops the handler and returns that status with a JSON body. Raise it for expected failures: a missing item, a bad state, an upstream provider that’s down.

10. Dependencies hold the shared plumbing. A dependency is a function FastAPI calls for you before your endpoint runs, declared as a parameter with Depends(get_db), usually through an Annotated alias so it’s written once. Shared logic, database sessions and auth live there instead of in every handler. Dependencies can be async def or plain def.

11. def versus async def endpoints. An async def endpoint runs on the event loop, so anything blocking inside it stalls every request (#03, point 12). A plain def endpoint runs in a threadpool instead, so blocking calls there are survivable. Rule of thumb: async def with async clients, plain def when your libraries are synchronous.

12. Settings come from the environment, validated. A pydantic-settings BaseSettings class reads environment variables (case-insensitively) and validates their types when it’s created. A missing API key or TIMEOUT_S=abc fails right there, not halfway through a request that uses it: at startup if you create the settings at import time, or on first use if you serve them the way the FastAPI docs do, from an @lru_cache ‘d get_settings() dependency. That dependency also lets tests swap settings with app.dependency_overrides.

Make async def fetch(i) that sleeps for a random 0–2 seconds and sometimes raises. Run 20 of them three ways: gather, gather(return_exceptions=True) and a TaskGroup. Then add Semaphore(5) and a 1-second timeout, and compare what you get back each time. For the FastAPI half, give a toy notes app a NoteOut model without an internal owner_id field, return an object that has one, and check that it doesn’t appear in the response.

Fan out with a limit and a deadline. Let FastAPI validate both directions, and keep shared plumbing in dependencies.

Where this fits in my roadmap: this is the second half of my foundations block (async in practice, plus FastAPI and Pydantic). In my own study, I’m building that block’s small LLM endpoint myself, with real timeouts and error handling. So this post sticks to the concepts and toy examples, not my solution. Next comes testing (pytest, plus mocking the provider), and then structured outputs, where the Pydantic schema from #03 becomes the contract the model must follow.

Over to you: gather(return_exceptions=True) or TaskGroup: which do you reach for by default, and why?

Originally published on The Path to AI FDE on Substack. Subscribe for the next issue: https://dhanushkumarsivaji.substack.com/subscribe