← FastAPI: practical and tested APIs Progress & review ↗

Typorithm · medium

Bound concurrent asynchronous requests

pythonfastapiindependent-practice

Saved on this device

Full lesson · 4 learning steps Open any exercise for focused practice

Step 1 of 4 · intro card

Bound concurrent asynchronous requests

Concept and contract

Use await, an in-process ASGI transport and an explicit concurrency budget.

An async endpoint can yield control while waiting, allowing other work to progress. That does not make unlimited concurrency safe, and it does not make blocking CPU work disappear. Use a bounded example to distinguish overlap, ordering and resource limits.

The fixture sends several requests through HTTPX ASGITransport into a local FastAPI app. No socket or public service is contacted. A semaphore permits at most two in-flight client operations, and a deterministic zero-duration yield lets the example observe overlap without relying on an unreliable speed benchmark.

Mechanism and worked example

Each endpoint increments an active counter, yields control and then returns a transformed identifier.

Each endpoint increments an active counter, yields control and then returns a transformed identifier. A finally block decrements the counter. The client wraps each request in the semaphore and awaits asyncio.gather, which returns results in input order even when work overlaps.

The helper reports returned values and the peak active count. Those observations show both the data contract and the concurrency budget. The synchronous outer helper uses asyncio.run only as a standalone lab entry point; calling it inside an already running event loop would require a different integration.

Complete worked example

python

import asyncio
import httpx
from fastapi import FastAPI

def inspect_example(values):
    app = FastAPI()
    active = peak = 0
    @app.get("/double/{value}")
    async def double(value: int):
        nonlocal active, peak
        active += 1
        peak = max(peak, active)
        try:
            await asyncio.sleep(0)
            return {"value": value * 2}
        finally:
            active -= 1
    async def run():
        budget = asyncio.Semaphore(2)
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://lesson.test") as client:
            async def one(value):
                async with budget:
                    response = await client.get(f"/double/{value}")
                    response.raise_for_status()
                    return response.json()["value"]
            return await asyncio.gather(*(one(value) for value in values))
    results = asyncio.run(run())
    return {"values": results, "peak": peak}

Reasoning: Each endpoint increments an active counter, yields control and then returns a transformed identifier. A finally block decrements the counter. The client wraps each request in the semaphore and awaits asyncio.gather, which returns results in input order even when work overlaps. The helper reports returned values and the peak active count. Those observations show both the data contract and the concurrency budget. The synchronous outer helper uses asyncio.run only as a standalone lab entry point; calling it inside an already running event loop would require a different integration.

Focused decision: use the complete example for context

python

budget = asyncio.Semaphore(2)

async def one(value):
    async with budget:
        response = await client.get(f"/double/{value}")
        response.raise_for_status()
        return response.json()["value"]

results = await asyncio.gather(*(one(value) for value in values))

Focused decision: Reconstruct this focused decision after the warm-up. The complete imports, model or application, and test client remain in the worked example. Typing records practice; only the downloadable computer lab executes Python.

Counterexamples and common mistakes

The counterexample sorts the results in reverse after gather.

The counterexample sorts the results in reverse after gather. All requests can succeed and the concurrency bound can remain correct, yet the output order violates the expected contract. Concurrent completion order and the requested result ordering are separate decisions.

ASGITransport does not automatically drive application lifespan. This particular fixture has no lifespan resources; a real async test needing startup must enter the appropriate lifespan manager explicitly. Do not infer readiness from the fact that an ASGI request returned a response.

Controlled mistake: observe the actual wrong behavior

python

import asyncio
import httpx
from fastapi import FastAPI

def inspect_example(values):
    app = FastAPI()
    active = peak = 0
    @app.get("/double/{value}")
    async def double(value: int):
        nonlocal active, peak
        active += 1
        peak = max(peak, active)
        try:
            await asyncio.sleep(0)
            return {"value": value * 2}
        finally:
            active -= 1
    async def run():
        budget = asyncio.Semaphore(2)
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://lesson.test") as client:
            async def one(value):
                async with budget:
                    response = await client.get(f"/double/{value}")
                    response.raise_for_status()
                    return response.json()["value"]
            return await asyncio.gather(*(one(value) for value in values))
    results = asyncio.run(run())
    return {"values": sorted(results, reverse=True), "peak": peak}

Reasoning: The counterexample sorts the results in reverse after gather. All requests can succeed and the concurrency bound can remain correct, yet the output order violates the expected contract. Concurrent completion order and the requested result ordering are separate decisions. ASGITransport does not automatically drive application lifespan. This particular fixture has no lifespan resources; a real async test needing startup must enter the appropriate lifespan manager explicitly. Do not infer readiness from the fact that an ASGI request returned a response.

Independent practice: one bounded change

Reduce the concurrency budget from two to one while preserving input-order results.

Reduce the concurrency budget from two to one while preserving input-order results. The peak should now be one for nonempty inputs and zero for no requests. Do not remove await or replace the work with constants merely to make the counter smaller.

Check three values, one value and an empty input. The transformed results must be unchanged. This small policy change is useful practice for adjusting resource pressure without rewriting business logic or claiming that serialized execution is universally the fastest choice.

Step 1: Serialize the in-flight client operations with a budget of one while preserving their results and order. Step 2: Retain the working imports and application structure. Change the stated policy, run the supplied practice cases, then explain the newly accepted or rejected boundary.

Practice starter: solve before reading the solution

python

import asyncio
import httpx
from fastapi import FastAPI

def inspect_example(values):
    app = FastAPI()
    active = peak = 0
    @app.get("/double/{value}")
    async def double(value: int):
        nonlocal active, peak
        active += 1
        peak = max(peak, active)
        try:
            await asyncio.sleep(0)
            return {"value": value * 2}
        finally:
            active -= 1
    async def run():
        budget = asyncio.Semaphore(2)
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://lesson.test") as client:
            async def one(value):
                async with budget:
                    response = await client.get(f"/double/{value}")
                    response.raise_for_status()
                    return response.json()["value"]
            return await asyncio.gather(*(one(value) for value in values))
    results = asyncio.run(run())
    return {"values": results, "peak": peak}

# Independent task: Serialize the in-flight client operations with a budget of one while preserving their results and order.

Reasoning: Serialize the in-flight client operations with a budget of one while preserving their results and order.

Hint 1: Find the one field or branch that owns this policy.
Hint 2: Compare the first changed acceptance case with the retained baseline.

Solution, tests and trade-offs

The solution changes the semaphore capacity.

The solution changes the semaphore capacity. Each operation still awaits the local HTTP request and releases the semaphore through its context manager. The endpoint finally block maintains the active count. The fixture checks exact data as well as the new peak.

A production service also needs deadlines, cancellation handling and limits at upstream and downstream boundaries. A semaphore is not a durable queue, retry policy or distributed rate limiter. Blocking libraries inside async def can still block the event loop; use suitable asynchronous libraries or an intentional worker boundary.

Practice solution and acceptance checks

python

import asyncio
import httpx
from fastapi import FastAPI

def inspect_example(values):
    app = FastAPI()
    active = peak = 0
    @app.get("/double/{value}")
    async def double(value: int):
        nonlocal active, peak
        active += 1
        peak = max(peak, active)
        try:
            await asyncio.sleep(0)
            return {"value": value * 2}
        finally:
            active -= 1
    async def run():
        budget = asyncio.Semaphore(1)
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://lesson.test") as client:
            async def one(value):
                async with budget:
                    response = await client.get(f"/double/{value}")
                    response.raise_for_status()
                    return response.json()["value"]
            return await asyncio.gather(*(one(value) for value in values))
    results = asyncio.run(run())
    return {"values": results, "peak": peak}

Reasoning: The solution changes the semaphore capacity. Each operation still awaits the local HTTP request and releases the semaphore through its context manager. The endpoint finally block maintains the active count. The fixture checks exact data as well as the new peak. A production service also needs deadlines, cancellation handling and limits at upstream and downstream boundaries. A semaphore is not a durable queue, retry policy or distributed rate limiter. Blocking libraries inside async def can still block the event loop; use suitable asynchronous libraries or an intentional worker boundary.

Interview reasoning and next step

Explain why gather preserves the input order here and why the semaphore is separate from that property.

Explain why gather preserves the input order here and why the semaphore is separate from that property. Describe the missing lifespan responsibility of ASGITransport. These distinctions prevent plausible-looking async tests from proving less than their names suggest.

Next, inspect host and browser-origin middleware. Passing an asynchronous request successfully does not establish which hosts are accepted or which browser origins can read a response. Network policy and application authorization remain separate.

Optional stretch — after the core task works: Add one independently reasoned boundary case and explain its expected value before running it.

Computer practice: https://applydjinn.com/typorithm/python-api-lab-20261105.zip. Use an isolated Python environment and the pinned dependencies. Run verify.py --track fastapi, then edit the working exercise baseline and run --practice fa-async. The reference suite rejects meaningful counterexamples and checks values, types and unchanged inputs. All data is owned and all HTTP clients are in-process. No production database, feed, account or paid service is used.

Use await, an in-process ASGI transport and an explicit concurrency budget.