fastapi 0.141.1: Manage FastAPI dependency injection lifecycles, request caching with Depends, background task ordering, yield exception handling, and response status/model interaction

# FastAPI Dependency Injection Lifecycle, Yield Exceptions, and Response Mechanics

## Search Results
`search_known_solution` returned existing sample `sha256:8bcaa6f98cea6edf681d2b0f76a18856106de62aab70009b1d22a5a50abb1257` (*"Test a FastAPI app with TestClient and no network, and assert the validation error bodies it returns"*), which covered 422 validation error payloads, direct function parameter coercion bypass, response_model filtering, lifespan context managers, and Starlette 1.6's httpx2 fallback.

## What a Model Would Have Written
A model would write standard `try ... except Exception: db.rollback()` inside a generator dependency without re-raising `raise`, and return `JSONResponse(content=...)` assuming `@app.post(..., status_code=201, response_model=...)` still applies.

## How the Wrong Version Fails
- Swallowing `Exception` in a `yield` dependency fails **loudly** at runtime with `fastapi.exceptions.FastAPIError: Response not awaited...` because `HTTPException` inherits from `Exception` and is swallowed before reaching FastAPI's error handlers.
- Returning `JSONResponse(...)` directly fails **silently with a green test/200 OK** that completely ignores the declared `status_code=201` and bypasses `response_model`, quietly leaking sensitive model attributes into the HTTP payload.

{"case":{"caseId":"case:sha256:2ca86f554e8f9178023c18c420dc262805056effe9b5fdf59232dd4f223ba96d","contract":["assert Depends(func) caches within request scope so sibling and child dependencies share the same instance","assert Depends(func, use_cache=False) bypasses request cache so each dependency receives a fresh instance","assert generator dependencies with yield unwind teardown in reverse LIFO order after the handler returns","assert BackgroundTasks execute before generator dependencies exit so database sessions remain open during background jobs","assert route HTTPException is forwarded into yield dependencies where except Exception catches it","assert catching HTTPException in a yield dependency without re-raising triggers FastAPIError response not awaited","assert app.dependency_overrides swaps dependencies for TestClient and clear restores defaults","assert direct endpoint function calls receive raw Depends objects rather than resolved dependency values","assert returning JSONResponse directly bypasses response_model and overrides status_code with 200","assert injecting response: Response parameter modifies status code and headers while preserving response_model filtering"],"goal":"Manage FastAPI dependency injection lifecycles, request caching with Depends, background task ordering, yield exception handling, and response status/model interaction","kind":"HOW","packages":["pkg:pypi/fastapi@0.141.1","pkg:pypi/starlette@1.6.0","pkg:pypi/pydantic@2.13.4","pkg:pypi/httpx2@2.10.0"],"schemaVersion":1,"symbols":["fastapi.FastAPI","fastapi.Depends","fastapi.HTTPException","fastapi.BackgroundTasks","fastapi.Response","fastapi.testclient.TestClient","fastapi.exceptions.FastAPIError"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","ecosystem":"pypi","executionContext":"python","language":"python","os":"linux","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/fastapi@0.141.1","pkg:pypi/starlette@1.6.0","pkg:pypi/pydantic@2.13.4","pkg:pypi/httpx2@2.10.0"],"schemaVersion":1,"symbols":["fastapi.FastAPI","fastapi.Depends","fastapi.HTTPException","fastapi.BackgroundTasks","fastapi.Response","fastapi.testclient.TestClient","fastapi.exceptions.FastAPIError"],"verifierAdapter":"python@1"}

fastapi==0.141.1
starlette==1.6.0
annotated-doc==0.0.5
pydantic==2.13.4
pydantic-core==2.46.4
typing-extensions==4.16.0
typing-inspection==0.4.4
annotated-types==0.8.0
anyio==4.14.2
idna==3.18
httpx2==2.10.0
httpcore2==2.10.0
truststore==0.10.4
httpx==0.28.1
httpcore==1.0.9
h11==0.16.0
certifi==2026.7.22

"""Package root for src."""

"""FastAPI application demonstrating dependency injection lifecycle, caching, and response mechanics."""

import itertools
from typing import Generator

from fastapi import BackgroundTasks, Depends, FastAPI, HTTPException, Response, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

seq_id = itertools.count(1)
EVENT_LOG: list[str] = []


# --- 1. Dependency Caching: use_cache=True vs use_cache=False ---
def get_session_id() -> int:
    return next(seq_id)


def service_with_cached_session(s_id: int = Depends(get_session_id)) -> dict[str, int]:
    return {"service_session": s_id}


@app.get("/cached-deps")
def cached_deps_endpoint(
    direct: int = Depends(get_session_id),
    srv: dict[str, int] = Depends(service_with_cached_session),
) -> dict[str, int]:
    return {"direct": direct, "srv": srv["service_session"]}


def get_uncached_session_id() -> int:
    return next(seq_id)


def service_with_uncached_session(
    s_id: int = Depends(get_uncached_session_id, use_cache=False),
) -> dict[str, int]:
    return {"service_session": s_id}


@app.get("/uncached-deps")
def uncached_deps_endpoint(
    direct: int = Depends(get_uncached_session_id, use_cache=False),
    srv: dict[str, int] = Depends(service_with_uncached_session),
) -> dict[str, int]:
    return {"direct": direct, "srv": srv["service_session"]}


# --- 2. Yield Lifecycle + BackgroundTasks ---
def resource_a() -> Generator[str, None, None]:
    EVENT_LOG.append("acquire:A")
    try:
        yield "A"
    finally:
        EVENT_LOG.append("release:A")


def resource_b(a: str = Depends(resource_a)) -> Generator[str, None, None]:
    EVENT_LOG.append("acquire:B")
    try:
        yield f"B_with_{a}"
    finally:
        EVENT_LOG.append("release:B")


def background_worker_task() -> None:
    EVENT_LOG.append("background:worker_run")


@app.get("/lifecycle")
def lifecycle_endpoint(
    bg: BackgroundTasks, b: str = Depends(resource_b)
) -> dict[str, str]:
    EVENT_LOG.append("handler:run")
    bg.add_task(background_worker_task)
    return {"resource": b}


# --- 3. Yield Exception Propagation & Swallowing Trap ---
def safe_db_session() -> Generator[str, None, None]:
    EVENT_LOG.append("db:open")
    try:
        yield "db_session_active"
    except Exception as exc:
        EVENT_LOG.append(f"db:caught:{type(exc).__name__}")
        raise  # Must re-raise so FastAPI handles the exception!
    finally:
        EVENT_LOG.append("db:close")


def broken_swallowing_db_session() -> Generator[str, None, None]:
    EVENT_LOG.append("broken_db:open")
    try:
        yield "broken_session_active"
    except Exception as exc:
        EVENT_LOG.append(f"broken_db:swallowed:{type(exc).__name__}")
        # Trapping error: catches Exception (which includes HTTPException) and fails to re-raise!
    finally:
        EVENT_LOG.append("broken_db:close")


@app.get("/db-error-safe")
def db_error_safe_endpoint(db: str = Depends(safe_db_session)) -> dict[str, str]:
    EVENT_LOG.append("handler:fail")
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN, detail="forbidden database operation"
    )


@app.get("/db-error-swallowed")
def db_error_swallowed_endpoint(
    db: str = Depends(broken_swallowing_db_session),
) -> dict[str, str]:
    EVENT_LOG.append("handler:fail")
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN, detail="forbidden database operation"
    )


# --- 4. Direct Response Return vs Response Parameter Injection ---
class UserOut(BaseModel):
    username: str


@app.post(
    "/response-injection",
    status_code=status.HTTP_201_CREATED,
    response_model=UserOut,
)
def response_injection_endpoint(response: Response) -> dict[str, str]:
    # Mutating the injected Response parameter modifies status and headers while keeping response_model active
    response.status_code = status.HTTP_202_ACCEPTED
    response.headers["x-custom-status"] = "injected"
    return {"username": "ada", "password_hash": "secret_abc123"}


@app.post(
    "/response-direct-return",
    status_code=status.HTTP_201_CREATED,
    response_model=UserOut,
)
def response_direct_return_endpoint() -> JSONResponse:
    # Returning JSONResponse directly overrides declared status_code with default 200 and bypasses response_model filtering
    return JSONResponse(
        content={"username": "ada", "password_hash": "secret_abc123"}
    )


# --- 5. Direct Endpoint Call Demonstration ---
def get_standalone_dep() -> dict[str, str]:
    return {"status": "connected"}


@app.get("/standalone")
def standalone_endpoint(
    dep: dict[str, str] = Depends(get_standalone_dep),
) -> dict[str, str]:
    return dep

import sys
from pathlib import Path

# Insert project root for module imports
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import fastapi
from fastapi import __version__ as fastapi_version
from fastapi.exceptions import FastAPIError
from fastapi.params import Depends as DependsType
from fastapi.testclient import TestClient

from src.app import (
    EVENT_LOG,
    app,
    get_session_id,
    standalone_endpoint,
)

client = TestClient(app)

# 1. Dependency Caching within Request Scope (use_cache=True)
# Sibling dependencies and route parameters requesting the same callable share the cached instance.
req1 = client.get("/cached-deps")
assert req1.status_code == 200
data1 = req1.json()
assert data1["direct"] == data1["srv"], "Default use_cache=True must share the instance within a single request"

req2 = client.get("/cached-deps")
assert req2.status_code == 200
data2 = req2.json()
assert data2["direct"] == data2["srv"]
assert data2["direct"] != data1["direct"], "Cache must not leak across separate requests"

# 2. Uncached Dependencies (use_cache=False)
# When use_cache=False is specified, each evaluation invokes the factory anew.
req_uncached = client.get("/uncached-deps")
assert req_uncached.status_code == 200
data_uncached = req_uncached.json()
assert data_uncached["direct"] != data_uncached["srv"], "use_cache=False must not share instances in the same request"

# 3. Generator (yield) Lifecycle and BackgroundTasks Execution Order
# Setup is FIFO, background tasks execute before generator teardown, and teardown is LIFO (reverse order).
EVENT_LOG.clear()
res_lifecycle = client.get("/lifecycle")
assert res_lifecycle.status_code == 200
assert res_lifecycle.json() == {"resource": "B_with_A"}
assert EVENT_LOG == [
    "acquire:A",
    "acquire:B",
    "handler:run",
    "background:worker_run",
    "release:B",
    "release:A",
], "Teardown must run in reverse LIFO order and keep resources open during background tasks"

# 4. Exception Forwarding into Generator Dependencies (re-raised)
# When a route raises HTTPException, it is thrown into active generators. Re-raising allows FastAPI to build the HTTP error response.
EVENT_LOG.clear()
res_safe = client.get("/db-error-safe")
assert res_safe.status_code == 403
assert res_safe.json() == {"detail": "forbidden database operation"}
assert EVENT_LOG == [
    "db:open",
    "handler:fail",
    "db:caught:HTTPException",
    "db:close",
], "Generator must catch the HTTPException, execute finally, and allow FastAPI to return 403"

# 5. Swallowed Exception Trap in Generator Dependencies
# If a generator catches Exception (which matches HTTPException) and fails to re-raise, FastAPI raises FastAPIError.
EVENT_LOG.clear()
swallowed_error_raised = False
try:
    client.get("/db-error-swallowed")
except FastAPIError as exc:
    swallowed_error_raised = True
    assert "Response not awaited" in str(exc)
assert swallowed_error_raised, "Swallowing HTTPException in a yield dependency must raise FastAPIError"
assert EVENT_LOG == [
    "broken_db:open",
    "handler:fail",
    "broken_db:swallowed:HTTPException",
    "broken_db:close",
]

# 6. Dependency Overrides
# app.dependency_overrides swaps dependencies during test execution.
app.dependency_overrides[get_session_id] = lambda: 8888
res_override = client.get("/cached-deps").json()
assert res_override["direct"] == 8888
assert res_override["srv"] == 8888
app.dependency_overrides.clear()
res_restored = client.get("/cached-deps").json()
assert res_restored["direct"] != 8888

# 7. Direct Function Calling vs ASGI Pipeline
# Direct endpoint calls bypass dependency injection, leaving default parameters as raw Depends objects.
direct_result = standalone_endpoint()
assert isinstance(direct_result, DependsType), "Direct call without arguments must return raw Depends descriptor"
mock_result = standalone_endpoint(dep={"mocked": "value"})
assert mock_result == {"mocked": "value"}, "Direct call with manual arguments must accept passed values"

# 8. Response Parameter Injection vs Direct JSONResponse Return
# Modifying injected Response alters status/headers and preserves response_model filtering.
res_injected = client.post("/response-injection")
assert res_injected.status_code == 202
assert res_injected.headers["x-custom-status"] == "injected"
assert res_injected.json() == {"username": "ada"}, "response_model must filter out secret fields"

# Returning JSONResponse directly overrides declared status_code (201 -> 200) and completely bypasses response_model.
res_raw = client.post("/response-direct-return")
assert res_raw.status_code == 200, "JSONResponse default status 200 overrides decorator status_code 201"
assert res_raw.json() == {"username": "ada", "password_hash": "secret_abc123"}, "JSONResponse return bypasses response_model"

print(f"Contract verified successfully against fastapi {fastapi_version}")