← FastAPI: practical and tested APIs Progress & review ↗

Typorithm · medium

Choose dependency cleanup scope for streaming

pythonfastapiindependent-practice

Saved on this device

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

Step 1 of 4 · intro card

Choose dependency cleanup scope for streaming

Concept and contract

Distinguish request cleanup after the response from function cleanup before it.

A dependency with yield owns work before and after the value it supplies. The cleanup point matters when the endpoint returns a streaming response whose content is produced later. A resource closed before content generation cannot safely support that generation.

Current FastAPI provides explicit dependency scope. The request scope keeps the dependency through response handling; function scope finishes it after the endpoint returns and before the response is sent. The lesson tests this actual ordering rather than inferring it from the word request.

Mechanism and worked example

The synthetic resource has one open flag.

The synthetic resource has one open flag. Its dependency records acquisition, yields the resource and closes it in finally. The streaming generator records when it runs and emits open or closed according to the resource state at that instant.

A local TestClient reads the response fully before the helper inspects the trace. With scope=request, acquisition precedes streaming and release follows it. The test observes a concrete sequence and body. It does not rely on wall-clock timing or a real connection pool.

Complete worked example

python

from typing import Annotated
from fastapi import Depends, FastAPI
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient

def inspect_example(requests):
    app = FastAPI()
    events = []
    def resource():
        state = {"open": True}
        events.append("acquire")
        try:
            yield state
        finally:
            state["open"] = False
            events.append("release")
    @app.get("/stream")
    def read(state: Annotated[dict, Depends(resource, scope="request")]):
        def chunks():
            events.append("stream")
            yield "open" if state["open"] else "closed"
        return StreamingResponse(chunks(), media_type="text/plain")
    with TestClient(app) as client:
        bodies = [client.get("/stream").text for _ in range(requests)]
    return {"bodies": bodies, "events": events}

Reasoning: The synthetic resource has one open flag. Its dependency records acquisition, yields the resource and closes it in finally. The streaming generator records when it runs and emits open or closed according to the resource state at that instant. A local TestClient reads the response fully before the helper inspects the trace. With scope=request, acquisition precedes streaming and release follows it. The test observes a concrete sequence and body. It does not rely on wall-clock timing or a real connection pool.

Focused decision: use the complete example for context

python

def resource():
    state = {"open": True}
    try:
        yield state
    finally:
        state["open"] = False

state: Annotated[dict, Depends(resource, scope="request")]

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 selects scope=function while leaving a stream that still inspects the resource.

The counterexample selects scope=function while leaving a stream that still inspects the resource. Cleanup then precedes streaming, so the body reports closed. An endpoint function returning without error is insufficient evidence that deferred response work has usable dependencies.

Do not hold a scarce database connection through an arbitrarily slow stream unless it is actually needed. Often the better design is to materialize bounded response data before returning, or acquire a stream-specific resource with a clear limit. Scope selection should follow ownership, not just whichever value makes one test pass.

Controlled mistake: observe the actual wrong behavior

python

from typing import Annotated
from fastapi import Depends, FastAPI
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient

def inspect_example(requests):
    app = FastAPI()
    events = []
    def resource():
        state = {"open": True}
        events.append("acquire")
        try:
            yield state
        finally:
            state["open"] = False
            events.append("release")
    @app.get("/stream")
    def read(state: Annotated[dict, Depends(resource, scope="function")]):
        def chunks():
            events.append("stream")
            yield "open" if state["open"] else "closed"
        return StreamingResponse(chunks(), media_type="text/plain")
    with TestClient(app) as client:
        bodies = [client.get("/stream").text for _ in range(requests)]
    return {"bodies": bodies, "events": events}

Reasoning: The counterexample selects scope=function while leaving a stream that still inspects the resource. Cleanup then precedes streaming, so the body reports closed. An endpoint function returning without error is insufficient evidence that deferred response work has usable dependencies. Do not hold a scarce database connection through an arbitrarily slow stream unless it is actually needed. Often the better design is to materialize bounded response data before returning, or acquire a stream-specific resource with a clear limit. Scope selection should follow ownership, not just whichever value makes one test pass.

Independent practice: one bounded change

Deliberately switch to function scope and update your predicted trace: acquire, release, stream.

Deliberately switch to function scope and update your predicted trace: acquire, release, stream. The body should now say closed. This task makes the alternative policy observable; it is not a recommendation to stream from a closed database session.

Check one and two responses plus an empty request loop. Keep finally release intact. In a real design, pair function scope with a response that no longer needs the resource. Explain that redesign before choosing the shorter lifetime for an actual streaming endpoint.

Step 1: Observe function-scope release before the stream and explain why the stream must not need that resource. 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

from typing import Annotated
from fastapi import Depends, FastAPI
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient

def inspect_example(requests):
    app = FastAPI()
    events = []
    def resource():
        state = {"open": True}
        events.append("acquire")
        try:
            yield state
        finally:
            state["open"] = False
            events.append("release")
    @app.get("/stream")
    def read(state: Annotated[dict, Depends(resource, scope="request")]):
        def chunks():
            events.append("stream")
            yield "open" if state["open"] else "closed"
        return StreamingResponse(chunks(), media_type="text/plain")
    with TestClient(app) as client:
        bodies = [client.get("/stream").text for _ in range(requests)]
    return {"bodies": bodies, "events": events}

# Independent task: Observe function-scope release before the stream and explain why the stream must not need that resource.

Reasoning: Observe function-scope release before the stream and explain why the stream must not need that resource.

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 only the declared dependency scope.

The solution changes only the declared dependency scope. The helper demonstrates the changed cleanup order for each request, and the fixture still isolates resources between requests. The zero-request case performs no dependency acquisition, unlike application lifespan entry.

Cleanup errors are operational failures, and exception handling around yield must preserve appropriate propagation. Suppressing every exception can conceal a failed request or release. This example has a deliberately simple cleanup path so ordering remains the central learning target.

Practice solution and acceptance checks

python

from typing import Annotated
from fastapi import Depends, FastAPI
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient

def inspect_example(requests):
    app = FastAPI()
    events = []
    def resource():
        state = {"open": True}
        events.append("acquire")
        try:
            yield state
        finally:
            state["open"] = False
            events.append("release")
    @app.get("/stream")
    def read(state: Annotated[dict, Depends(resource, scope="function")]):
        def chunks():
            events.append("stream")
            yield "open" if state["open"] else "closed"
        return StreamingResponse(chunks(), media_type="text/plain")
    with TestClient(app) as client:
        bodies = [client.get("/stream").text for _ in range(requests)]
    return {"bodies": bodies, "events": events}

Reasoning: The solution changes only the declared dependency scope. The helper demonstrates the changed cleanup order for each request, and the fixture still isolates resources between requests. The zero-request case performs no dependency acquisition, unlike application lifespan entry. Cleanup errors are operational failures, and exception handling around yield must preserve appropriate propagation. Suppressing every exception can conceal a failed request or release. This example has a deliberately simple cleanup path so ordering remains the central learning target.

Interview reasoning and next step

Explain the difference between endpoint return, response production and request completion.

Explain the difference between endpoint return, response production and request completion. Then identify which scope keeps the resource available to this generator. A precise answer can point to the trace rather than treating all yield dependencies as equivalent context managers.

Next, test asynchronous application calls with an in-process transport. Asynchrony allows overlapping waits but does not automatically bound concurrency or run application lifespan. Those responsibilities need explicit tests too.

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-yield. 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.

Distinguish request cleanup after the response from function cleanup before it.