starlette 1.4.1: requests.Request

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 starlette.requests.Request in pkg:pypi/starlette@1.4.1
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/starlette@1.4.1
Demonstrate these symbols/APIs:
  - starlette.requests.Request

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:4e4ae1c2c80549337252e65996608e0cc4ff187c4255efbbeb0adbb3dda3150a","contract":["Request extracts HTTP method, headers, cookies, and query parameters from ASGI connection scope","Request reads and parses JSON payload asynchronously from request body stream","Request provides client connection details and request-scoped state storage"],"goal":"verify starlette.requests.Request in pkg:pypi/starlette@1.4.1","kind":"HOW","packages":["pkg:pypi/starlette@1.4.1"],"schemaVersion":1,"symbols":["starlette.requests.Request"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/starlette@1.4.1"],"schemaVersion":1,"subject":"pkg:pypi/starlette@1.4.1","symbols":["starlette.requests.Request"],"verifierAdapter":"python@1"}

anyio==4.14.2
idna==3.19
starlette==1.4.1
typing-extensions==4.16.0

"""Sample demonstrating HTTP request parsing, payload processing, and state handling with Starlette Request."""

from typing import Any, Dict
from starlette.requests import Request


def get_request_metadata(request: Request) -> Dict[str, Any]:
    """Extract connection metadata, headers, cookies, and query parameters from Request."""
    return {
        "method": request.method,
        "path": request.url.path,
        "query_params": dict(request.query_params),
        "headers": {
            "content_type": request.headers.get("content-type"),
            "request_id": request.headers.get("x-request-id"),
        },
        "cookies": dict(request.cookies),
    }


async def read_request_payload(request: Request) -> Dict[str, Any]:
    """Read raw payload bytes and parse JSON data asynchronously from Request."""
    body_bytes = await request.body()
    json_data = await request.json()
    return {
        "byte_count": len(body_bytes),
        "parsed": json_data,
    }


def manage_request_state(request: Request, key: str, value: Any) -> Any:
    """Store and retrieve arbitrary state attributes on request.state."""
    setattr(request.state, key, value)
    return getattr(request.state, key)

{
  "schemaVersion": 1,
  "goal": "verify starlette.requests.Request in pkg:pypi/starlette@1.4.1",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/starlette@1.4.1"
  ],
  "symbols": [
    "starlette.requests.Request"
  ]
}

"""Contract verification tests for starlette.requests.Request."""

import asyncio
import json
import os
import sys
from typing import Any, Dict, List, Optional, Tuple

# Ensure project root is on sys.path for local module imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from starlette.requests import Request
from sample import get_request_metadata, read_request_payload, manage_request_state


def create_asgi_http_scope(
    method: str = "GET",
    path: str = "/",
    query_string: bytes = b"",
    headers: Optional[List[Tuple[bytes, bytes]]] = None,
    client: Optional[Tuple[str, int]] = None,
) -> Dict[str, Any]:
    """Construct an ASGI HTTP connection scope dictionary for offline testing."""
    return {
        "type": "http",
        "http_version": "1.1",
        "method": method,
        "scheme": "http",
        "server": ("127.0.0.1", 80),
        "path": path,
        "raw_path": path.encode("latin-1"),
        "query_string": query_string,
        "headers": headers or [],
        "client": client or ("127.0.0.1", 50000),
    }


def test_request_metadata_extraction() -> None:
    """Assertion 1: Request extracts HTTP method, headers, cookies, and query parameters from ASGI connection scope."""
    scope = create_asgi_http_scope(
        method="GET",
        path="/api/search",
        query_string=b"q=starlette&page=1",
        headers=[
            (b"content-type", b"application/json"),
            (b"x-request-id", b"req-abc-123"),
            (b"cookie", b"session_id=sess-token-456"),
        ],
    )
    request = Request(scope)

    metadata = get_request_metadata(request)
    assert metadata["method"] == "GET"
    assert metadata["path"] == "/api/search"
    assert metadata["query_params"] == {"q": "starlette", "page": "1"}
    assert metadata["headers"]["content_type"] == "application/json"
    assert metadata["headers"]["request_id"] == "req-abc-123"
    assert metadata["cookies"] == {"session_id": "sess-token-456"}


def test_request_payload_and_json_parsing() -> None:
    """Assertion 2: Request reads and parses JSON payload asynchronously from request body stream."""
    payload = {"symbol": "Request", "status": "active", "items": [1, 2, 3]}
    payload_bytes = json.dumps(payload).encode("utf-8")

    async def receive() -> Dict[str, Any]:
        return {
            "type": "http.request",
            "body": payload_bytes,
            "more_body": False,
        }

    scope = create_asgi_http_scope(
        method="POST",
        path="/api/submit",
        headers=[(b"content-type", b"application/json")],
    )
    request = Request(scope, receive=receive)

    async def run_async_test() -> None:
        result = await read_request_payload(request)
        assert result["byte_count"] == len(payload_bytes)
        assert result["parsed"] == payload
        assert result["parsed"]["symbol"] == "Request"

    asyncio.run(run_async_test())


def test_request_client_and_state() -> None:
    """Assertion 3: Request provides client connection details and request-scoped state storage."""
    client_address = ("127.0.0.1", 54321)
    scope = create_asgi_http_scope(
        method="GET",
        path="/api/client",
        client=client_address,
    )
    request = Request(scope)

    assert request.client is not None
    assert request.client.host == "127.0.0.1"
    assert request.client.port == 54321

    stored_value = manage_request_state(request, "tenant_id", "tenant-001")
    assert stored_value == "tenant-001"
    assert request.state.tenant_id == "tenant-001"


def main() -> None:
    test_request_metadata_extraction()
    test_request_payload_and_json_parsing()
    test_request_client_and_state()
    print("Starlette Request contract verification passed.")


if __name__ == "__main__":
    main()