fastapi 0.141.1: Request, FastAPI, testclient.TestClient

# FastAPI Request Inspection, State Management, and Body Caching

## Search Results
`search_known_solution` returned `NO_SAFE_MATCH` for `fastapi.Request` inspection, headers/query parameters multi-value parsing, cookie extraction, middleware request state passing, and multi-read request body caching in `pkg:pypi/fastapi@0.141.1`. Existing samples for FastAPI covered APIRouter nesting, dependency injection, and Pydantic validation, but did not prove `fastapi.Request` ASGI scope handling, `.state`, `.client`, `.url_for`, and `.body()` / `.json()` stream caching semantics.

## Proofs
This sample verifies:
1. `fastapi.Request` is identical to `starlette.requests.Request` re-exported without modification.
2. Endpoint parameter typed `request: Request` receives the raw Request object without validation coercion.
3. Accessing `request.client`, `request.method`, `request.url`, `request.base_url`, and URL generation via `request.url_for`.
4. Header extraction with case insensitivity and multi-value query parameter retrieval with `.getlist()`.
5. Cookie extraction via `request.cookies`.
6. Attaching custom data to `request.state` in ASGI middleware and reading it inside endpoint handlers.
7. Multi-read capability of `await request.body()` and `await request.json()` without stream exhaustion errors.
8. Direct access to `request.scope` ASGI dictionary and constructing `Request` from a synthetic scope.

Clean-room public code sample — generation instructions

Write a brand-new, minimal, self-contained code sample in this clean-room directory.
Do not copy, paraphrase, or reference any existing project source. Work only from this spec.

A csx.json manifest scaffold already exists. Do not recreate it from memory. Preserve its case.goal, packages and symbols; fill its empty case.contract with exact assertions and correct its environment, commands and verifierAdapter for the files you generate.

Goal: verify fastapi.Request in pkg:pypi/fastapi@0.141.1
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/fastapi@0.141.1
Demonstrate these symbols/APIs:
  - fastapi.Request
Constraints:
  - executionContext: node
Required runtime conditions:
  - ecosystem: npm
  - language: javascript
  - moduleSystem: cjs
  - packageManager: npm@10.9.8
  - runtime: node@22.23.2

Rules:
  - One focused purpose; the smallest project that proves the goal.
  - Include a contract test (test/contract.*) that runs OFFLINE and exits 0 exactly when the goal behavior works.
  - Pin every dependency with a lockfile so resolution is reproducible.
  - No secrets, credentials, or tokens. No real URLs (only example.com or localhost). No absolute paths.
  - No personal names, emails, company names, or project identifiers of any kind.
  - No binaries and no generated output (node_modules, dist, target, venv, .git, .env).
  - Keep it under 200 files and 256KB packed.

{"case":{"caseId":"case:sha256:f5826d7c68fca8c059718d1a908e08bcc4cade884dd26c3c6f10b0b9f3523f87","constraints":{"executionContext":"python"},"contract":["assert fastapi.Request is starlette.requests.Request re-exported without modification","assert endpoint parameter typed as Request receives the raw Request object without validation coercion","assert request.method, request.url, and request.base_url provide normalized HTTP routing and scheme metadata","assert request.url_for resolves endpoint URL by name with path parameters","assert request.headers provides case-insensitive lookup for request headers","assert request.query_params supports get for single values and getlist for multi-value query strings","assert request.cookies parses cookie header into key-value pairs","assert request.state stores and propagates custom request-scoped attributes set by middleware to endpoint handlers","assert request.body and request.json parse payload and cache bytes across multiple reads without stream exhaustion","assert request.scope provides direct access to underlying ASGI scope dictionary","assert Request can be instantiated directly from an ASGI scope dictionary"],"goal":"verify fastapi.Request in pkg:pypi/fastapi@0.141.1","kind":"HOW","packages":["pkg:pypi/fastapi@0.141.1","pkg:pypi/starlette@1.6.0","pkg:pypi/pydantic@2.13.4","pkg:pypi/httpx@0.28.1"],"schemaVersion":1,"symbols":["fastapi.Request","fastapi.FastAPI","fastapi.testclient.TestClient"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","executionContext":"python","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","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/httpx@0.28.1"],"schemaVersion":1,"subject":"pkg:pypi/fastapi@0.141.1","symbols":["fastapi.Request","fastapi.FastAPI","fastapi.testclient.TestClient"],"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

{
  "schemaVersion": 1,
  "goal": "verify fastapi.Request in pkg:pypi/fastapi@0.141.1",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/fastapi@0.141.1"
  ],
  "symbols": [
    "fastapi.Request"
  ],
  "constraints": {
    "executionContext": "node"
  },
  "runtimeConditions": {
    "ecosystem": "npm",
    "language": "javascript",
    "moduleSystem": "cjs",
    "packageManager": "npm@10.9.8",
    "runtime": "node@22.23.2"
  }
}

"""Package root for src."""

"""FastAPI application demonstrating Request object inspection, state propagation, and body stream caching."""

from typing import Any
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.responses import JSONResponse, Response


class RequestStateMiddleware(BaseHTTPMiddleware):
    """Middleware attaching request-scoped metadata to request.state."""

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        # Attach request metadata to request.state for downstream handlers
        request.state.trace_id = request.headers.get("x-trace-id", "trace-default-123")
        request.state.custom_auth = {"user_id": 42, "role": "operator"}
        response = await call_next(request)
        response.headers["x-trace-id-echo"] = request.state.trace_id
        return response


app = FastAPI()
app.add_middleware(RequestStateMiddleware)


@app.get("/inspect/metadata")
async def inspect_metadata(request: Request) -> dict[str, Any]:
    """Inspect request client, method, url, base_url, headers, query_params, and cookies."""
    return {
        "client_host": request.client.host if request.client else None,
        "client_port": request.client.port if request.client else None,
        "method": request.method,
        "url_path": request.url.path,
        "url_scheme": request.url.scheme,
        "url_netloc": request.url.netloc,
        "url_str": str(request.url),
        "base_url_str": str(request.base_url),
        "header_custom": request.headers.get("x-custom-header"),
        "header_custom_upper": request.headers.get("X-CUSTOM-HEADER"),
        "query_single": request.query_params.get("filter"),
        "query_tags": request.query_params.getlist("tag"),
        "cookie_session": request.cookies.get("session_id"),
        "state_trace_id": getattr(request.state, "trace_id", None),
        "state_auth_user": getattr(request.state, "custom_auth", None),
        "scope_type": request.scope.get("type"),
        "scope_http_version": request.scope.get("http_version"),
    }


@app.post("/inspect/body-cache")
async def inspect_body_cache(request: Request) -> dict[str, Any]:
    """Demonstrate reading request.body() and request.json() repeatedly without stream exhaustion."""
    # First read as raw bytes
    raw_1 = await request.body()
    # Second read as parsed JSON (uses cached _body)
    json_1 = await request.json()
    # Third read as raw bytes again
    raw_2 = await request.body()
    # Fourth read as parsed JSON again
    json_2 = await request.json()

    return {
        "raw_equal": raw_1 == raw_2,
        "raw_bytes_len": len(raw_1),
        "json_equal": json_1 == json_2,
        "parsed_data": json_1,
    }


@app.get("/users/{user_id}", name="get_user_by_id")
async def get_user_by_id(user_id: int, request: Request) -> dict[str, Any]:
    """Endpoint using request.url_for to generate reverse URLs."""
    self_url = str(request.url_for("get_user_by_id", user_id=user_id))
    return {
        "user_id": user_id,
        "self_url": self_url,
    }

import sys
from pathlib import Path

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

import fastapi
import starlette.requests
from fastapi import Request
from fastapi.testclient import TestClient

from src.app import app

# --- 1. Verify re-export identity ---
assert Request is starlette.requests.Request, (
    "fastapi.Request must be starlette.requests.Request re-exported without modification"
)

client = TestClient(app)

# --- 2. Verify request metadata inspection: client, method, url, headers, query_params, cookies, state ---
headers = {
    "X-Custom-Header": "custom-value-123",
    "x-trace-id": "req-trace-999",
}
cookies = {
    "session_id": "sess-abc-xyz",
}

res = client.get(
    "/inspect/metadata?filter=active&tag=python&tag=fastapi",
    headers=headers,
    cookies=cookies,
)

assert res.status_code == 200
data = res.json()

# Endpoint parameter typed as Request receives the raw Request object
assert data["client_host"] == "testclient"
assert isinstance(data["client_port"], int)

# request.method, request.url, and request.base_url provide normalized HTTP routing and scheme metadata
assert data["method"] == "GET"
assert data["url_path"] == "/inspect/metadata"
assert data["url_scheme"] == "http"
assert data["url_netloc"] == "testserver"
assert data["url_str"] == "http://testserver/inspect/metadata?filter=active&tag=python&tag=fastapi"
assert data["base_url_str"] == "http://testserver/"

# request.headers provides case-insensitive lookup for request headers
assert data["header_custom"] == "custom-value-123"
assert data["header_custom_upper"] == "custom-value-123"

# request.query_params supports get for single values and getlist for multi-value query strings
assert data["query_single"] == "active"
assert data["query_tags"] == ["python", "fastapi"]

# request.cookies parses cookie header into key-value pairs
assert data["cookie_session"] == "sess-abc-xyz"

# request.state stores and propagates custom request-scoped attributes set by middleware to endpoint handlers
assert data["state_trace_id"] == "req-trace-999"
assert data["state_auth_user"] == {"user_id": 42, "role": "operator"}
assert res.headers["x-trace-id-echo"] == "req-trace-999"

# request.scope provides direct access to underlying ASGI scope dictionary
assert data["scope_type"] == "http"
assert data["scope_http_version"] in ("1.1", "2", "3")


# --- 3. Verify request.url_for reverses URL paths correctly ---
res_url = client.get("/users/42")
assert res_url.status_code == 200
assert res_url.json() == {
    "user_id": 42,
    "self_url": "http://testserver/users/42",
}


# --- 4. Verify request.body() and request.json() caching across multiple reads ---
payload = {"action": "sync", "count": 7, "flags": ["ready", "verified"]}
res_body = client.post("/inspect/body-cache", json=payload)
assert res_body.status_code == 200
body_data = res_body.json()
assert body_data["raw_equal"] is True
assert body_data["raw_bytes_len"] > 0
assert body_data["json_equal"] is True
assert body_data["parsed_data"] == payload


# --- 5. Verify direct instantiation of Request from synthetic ASGI scope ---
synthetic_scope = {
    "type": "http",
    "http_version": "1.1",
    "method": "POST",
    "path": "/synthetic/path",
    "raw_path": b"/synthetic/path",
    "query_string": b"foo=bar&num=1&num=2",
    "headers": [
        (b"host", b"localhost:8000"),
        (b"content-type", b"application/json"),
        (b"x-api-key", b"test-secret-key"),
        (b"cookie", b"user=alice; theme=dark"),
    ],
    "client": ("127.0.0.1", 54321),
    "server": ("localhost", 8000),
    "scheme": "http",
    "app": app,
    "state": {},
}

mock_request = Request(scope=synthetic_scope)
assert mock_request.method == "POST"
assert mock_request.client.host == "127.0.0.1"
assert mock_request.client.port == 54321
assert str(mock_request.url) == "http://localhost:8000/synthetic/path?foo=bar&num=1&num=2"
assert mock_request.headers.get("x-api-key") == "test-secret-key"
assert mock_request.headers.get("X-Api-Key") == "test-secret-key"
assert mock_request.query_params.get("foo") == "bar"
assert mock_request.query_params.getlist("num") == ["1", "2"]
assert mock_request.cookies.get("user") == "alice"
assert mock_request.cookies.get("theme") == "dark"
assert mock_request.scope["path"] == "/synthetic/path"

mock_request.state.tenant = "tenant-a"
assert mock_request.state.tenant == "tenant-a"

print(f"fastapi.Request contract verified successfully against fastapi {fastapi.__version__}")