starlette 1.6.0: types.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 starlette.types.Scope in pkg:pypi/starlette@1.6.0
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/starlette@1.6.0
Demonstrate these symbols/APIs:
  - starlette.types.Scope

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:b18d174dc2144da8cd4233d955c8bb3ce4ecdd483c3b00b2082171072e3342e7","contract":["Scope mapping provides ASGI connection metadata including method, path, and headers accessible via Request.scope","ASGI middleware inspects and mutates Scope state to propagate context to downstream handlers","Direct ASGI endpoint processes Scope mapping without Request abstraction"],"goal":"verify starlette.types.Scope in pkg:pypi/starlette@1.6.0","kind":"HOW","packages":["pkg:pypi/starlette@1.6.0"],"schemaVersion":1,"symbols":["starlette.types.Scope"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/starlette@1.6.0"],"schemaVersion":1,"subject":"pkg:pypi/starlette@1.6.0","symbols":["starlette.types.Scope"],"verifierAdapter":"python@1"}

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

"""Sample demonstrating ASGI Scope handling, inspection, and state enrichment in Starlette."""

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse, PlainTextResponse
from starlette.routing import Mount, Route
from starlette.types import ASGIApp, Receive, Scope, Send


class ScopeContextMiddleware:
    """ASGI middleware that inspects and mutates connection Scope metadata."""

    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] == "http":
            if "state" not in scope:
                scope["state"] = {}

            headers = dict(scope.get("headers", []))
            client_trace = headers.get(b"x-trace-id", b"trace-default").decode("latin-1")

            scope["state"]["trace_id"] = client_trace
            scope["state"]["handled_by_middleware"] = True

        await self.app(scope, receive, send)


async def inspect_scope_endpoint(request: Request) -> JSONResponse:
    """Endpoint that accesses request.scope to inspect connection and state metadata."""
    scope: Scope = request.scope
    trace_id = scope.get("state", {}).get("trace_id", "none")
    scope_type = scope.get("type", "")
    method = scope.get("method", "")
    path = scope.get("path", "")

    return JSONResponse(
        {
            "scope_type": scope_type,
            "method": method,
            "path": path,
            "trace_id": trace_id,
            "middleware_applied": scope.get("state", {}).get("handled_by_middleware", False),
        }
    )


async def raw_asgi_endpoint(scope: Scope, receive: Receive, send: Send) -> None:
    """Pure ASGI endpoint processing Scope directly without high-level wrappers."""
    assert scope["type"] == "http", "Expected HTTP scope"
    response = PlainTextResponse(
        f"Raw ASGI scope: type={scope.get('type')}, root_path={scope.get('root_path')}"
    )
    await response(scope, receive, send)


def create_app() -> ASGIApp:
    """Create and configure Starlette application wrapped with ScopeContextMiddleware."""
    routes = [
        Route("/inspect", endpoint=inspect_scope_endpoint, methods=["GET"]),
        Mount("/raw", app=raw_asgi_endpoint),
    ]
    starlette_app = Starlette(routes=routes)
    return ScopeContextMiddleware(starlette_app)


if __name__ == "__main__":
    app = create_app()
    print("Starlette Scope application created successfully.")

{
  "schemaVersion": 1,
  "goal": "verify starlette.types.Scope in pkg:pypi/starlette@1.6.0",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/starlette@1.6.0"
  ],
  "symbols": [
    "starlette.types.Scope"
  ]
}

"""Contract verification tests for starlette.types.Scope."""

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.types import ASGIApp, Receive, Scope, Send
from sample import ScopeContextMiddleware, create_app


async def invoke_asgi(
    app: ASGIApp,
    method: str,
    path: str,
    headers: Optional[List[Tuple[bytes, bytes]]] = None,
    initial_scope_state: Optional[Dict[str, Any]] = None,
) -> Tuple[int, Dict[bytes, bytes], bytes, Scope]:
    """Helper to invoke an ASGI application offline and capture the execution scope."""
    scope: Scope = {
        "type": "http",
        "asgi": {"version": "3.0", "spec_version": "2.3"},
        "http_version": "1.1",
        "method": method,
        "scheme": "http",
        "path": path,
        "raw_path": path.encode("latin-1"),
        "query_string": b"",
        "headers": headers or [],
        "client": ("127.0.0.1", 54321),
        "server": ("localhost", 80),
        "state": initial_scope_state.copy() if initial_scope_state is not None else {},
    }

    sent_messages: List[Dict[str, Any]] = []

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

    async def send(message: Dict[str, Any]) -> None:
        sent_messages.append(message)

    await app(scope, receive, send)

    status_code = 0
    resp_headers: Dict[bytes, bytes] = {}
    response_body = b""

    for msg in sent_messages:
        if msg["type"] == "http.response.start":
            status_code = msg["status"]
            resp_headers = dict(msg.get("headers", []))
        elif msg["type"] == "http.response.body":
            response_body += msg.get("body", b"")

    return status_code, resp_headers, response_body, scope


def test_scope_metadata_and_request_access() -> None:
    """Assertion 1: Scope mapping provides ASGI connection metadata including method, path, and headers accessible via Request.scope."""
    app = create_app()

    async def run_test() -> None:
        headers = [(b"x-trace-id", b"trace-test-456"), (b"host", b"localhost")]
        status, resp_headers, body, returned_scope = await invoke_asgi(
            app, "GET", "/inspect", headers=headers
        )

        assert status == 200, f"Expected 200, got {status}"
        data = json.loads(body.decode("utf-8"))
        assert data["scope_type"] == "http"
        assert data["method"] == "GET"
        assert data["path"] == "/inspect"
        assert data["trace_id"] == "trace-test-456"
        assert data["middleware_applied"] is True

        # Verify underlying Scope mapping properties
        assert returned_scope["type"] == "http"
        assert returned_scope["path"] == "/inspect"
        assert returned_scope["client"] == ("127.0.0.1", 54321)

    asyncio.run(run_test())


def test_scope_middleware_state_mutation() -> None:
    """Assertion 2: ASGI middleware inspects and mutates Scope state to propagate context to downstream handlers."""
    app = create_app()

    async def run_test() -> None:
        initial_state = {"seed_key": "seed_value"}
        status, _, body, returned_scope = await invoke_asgi(
            app, "GET", "/inspect", initial_scope_state=initial_state
        )

        assert status == 200, f"Expected 200, got {status}"
        data = json.loads(body.decode("utf-8"))
        assert data["trace_id"] == "trace-default"
        assert data["middleware_applied"] is True

        # Verify state mutation persists on Scope
        assert returned_scope["state"]["seed_key"] == "seed_value"
        assert returned_scope["state"]["handled_by_middleware"] is True
        assert returned_scope["state"]["trace_id"] == "trace-default"

    asyncio.run(run_test())


def test_raw_asgi_scope_handling() -> None:
    """Assertion 3: Direct ASGI endpoint processes Scope mapping without Request abstraction."""
    app = create_app()

    async def run_test() -> None:
        status, _, body, returned_scope = await invoke_asgi(app, "GET", "/raw/")
        assert status == 200, f"Expected 200, got {status}"
        assert body == b"Raw ASGI scope: type=http, root_path=/raw"
        assert returned_scope["state"]["handled_by_middleware"] is True

    asyncio.run(run_test())


def main() -> None:
    test_scope_metadata_and_request_access()
    test_scope_middleware_state_mutation()
    test_raw_asgi_scope_handling()
    print("Starlette Scope contract verification passed.")


if __name__ == "__main__":
    main()