starlette 1.6.0: types.Send

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

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:884a71d0dd84e6570b4d4ab75e99a24b0394e97469ab9b6088e648b4b6f8dc9f","contract":["starlette.types.Send is a type alias for an async callable accepting an ASGI Message and returning None","Wrapping the Send callable in ASGI middleware allows intercepting and modifying outgoing response headers","Direct invocations of the Send callable with more_body sequencing stream multiple payload chunks to the client","Invoking an ASGIApp with a collector conforming to Send records http.response.start and http.response.body in exact sequence","DirectSendResponse emits proper start and body events with more_body flags across multiple Send calls"],"goal":"verify starlette.types.Send in pkg:pypi/starlette@1.6.0","kind":"HOW","packages":["pkg:pypi/starlette@1.6.0"],"schemaVersion":1,"symbols":["starlette.types.Send"]},"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","runtimeVersion":"3.12","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/starlette@1.6.0"],"schemaVersion":1,"subject":"pkg:pypi/starlette@1.6.0","symbols":["starlette.types.Send"],"verifierAdapter":"python@1"}

starlette==1.6.0
anyio==4.14.2
idna==3.19
typing-extensions==4.16.0
httpx2==2.10.0
httpcore2==2.10.0
truststore==0.10.4

"""Sample demonstrating ASGI Send callable usage, middleware wrapping, and response streaming with starlette.types.Send."""

from typing import Any, Awaitable, Callable, List, Optional
from starlette.applications import Starlette
from starlette.datastructures import MutableHeaders
from starlette.responses import JSONResponse, PlainTextResponse, Response
from starlette.routing import Route
from starlette.types import ASGIApp, Message, Receive, Scope, Send


class HeaderInjectorSendMiddleware:
    """Pure ASGI middleware demonstrating wrapping an ASGI Send callable to modify outgoing headers."""

    def __init__(self, app: ASGIApp, header_name: str = "x-send-wrapper", header_value: str = "active") -> None:
        self.app = app
        self.header_name = header_name
        self.header_value = header_value

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        async def send_wrapper(message: Message) -> None:
            if message["type"] == "http.response.start":
                headers = MutableHeaders(scope=message)
                headers[self.header_name] = self.header_value
            await send(message)

        await self.app(scope, receive, send_wrapper)


class DirectSendResponse(Response):
    """Custom response class that coordinates send calls for chunked streaming."""

    def __init__(self, chunks: List[bytes], content_type: str = "text/plain") -> None:
        super().__init__()
        self.chunks = chunks
        self.media_type = content_type

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        headers = MutableHeaders()
        headers["content-type"] = self.media_type

        # Send response start message
        await send({
            "type": "http.response.start",
            "status": 200,
            "headers": headers.raw,
        })

        # Stream chunks through the Send callable
        for i, chunk in enumerate(self.chunks):
            is_last = (i == len(self.chunks) - 1)
            await send({
                "type": "http.response.body",
                "body": chunk,
                "more_body": not is_last,
            })


class SendEventCollector:
    """Helper to record all messages dispatched through a Send callable."""

    def __init__(self) -> None:
        self.messages: List[Message] = []

    async def send(self, message: Message) -> None:
        self.messages.append(dict(message))


async def ping_endpoint(request) -> JSONResponse:
    """Simple JSON endpoint returning status."""
    return JSONResponse({"status": "healthy", "symbol": "Send"})


async def stream_endpoint(request) -> DirectSendResponse:
    """Streaming endpoint delivering multiple data chunks."""
    chunks = [b"chunk-alpha;", b"chunk-beta;", b"chunk-gamma"]
    return DirectSendResponse(chunks=chunks)


def create_app(header_name: str = "x-send-wrapper", header_value: str = "active") -> ASGIApp:
    """Factory creating a Starlette application wrapped with Send middleware."""
    routes = [
        Route("/ping", ping_endpoint, methods=["GET"]),
        Route("/stream", stream_endpoint, methods=["GET"]),
    ]
    raw_app = Starlette(debug=False, routes=routes)
    return HeaderInjectorSendMiddleware(raw_app, header_name=header_name, header_value=header_value)


if __name__ == "__main__":
    app = create_app()
    print("Starlette Send sample application configured.")

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

"""Contract verification tests for starlette.types.Send in Starlette 1.6.0."""

import asyncio
from pathlib import Path
import sys
from typing import Awaitable, Callable, List

# Ensure project root is on sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import starlette.types
from starlette.testclient import TestClient
from starlette.types import ASGIApp, Message, Receive, Scope, Send

from sample import DirectSendResponse, HeaderInjectorSendMiddleware, SendEventCollector, create_app


def test_send_type_annotation_and_signature() -> None:
    """Assertion 1: starlette.types.Send is a type alias for an async callable accepting an ASGI Message and returning None."""
    # Verify starlette.types exposes Send
    assert hasattr(starlette.types, "Send")

    # Verify a conforming async callable satisfies the Send signature
    recorded: List[Message] = []

    async def custom_send(message: Message) -> None:
        recorded.append(message)

    # Validate callable execution via asyncio
    sample_message: Message = {
        "type": "http.response.start",
        "status": 200,
        "headers": [(b"content-type", b"text/plain")],
    }

    asyncio.run(custom_send(sample_message))
    assert len(recorded) == 1
    assert recorded[0]["status"] == 200


def test_send_wrapper_middleware_header_injection() -> None:
    """Assertion 2: Wrapping the Send callable in ASGI middleware allows intercepting and modifying outgoing response headers."""
    app: ASGIApp = create_app(header_name="x-custom-send-header", header_value="intercepted-v1")
    client = TestClient(app)

    resp = client.get("/ping")
    assert resp.status_code == 200
    assert resp.json() == {"status": "healthy", "symbol": "Send"}
    assert resp.headers["x-custom-send-header"] == "intercepted-v1"


def test_chunked_streaming_via_send_callable() -> None:
    """Assertion 3: Direct invocations of the Send callable with more_body sequencing stream multiple payload chunks to the client."""
    app: ASGIApp = create_app()
    client = TestClient(app)

    resp = client.get("/stream")
    assert resp.status_code == 200
    assert resp.text == "chunk-alpha;chunk-beta;chunk-gamma"
    assert resp.headers["x-send-wrapper"] == "active"


def test_direct_asgi_send_lifecycle() -> None:
    """Assertion 4: Invoking an ASGIApp with a collector conforming to Send records http.response.start and http.response.body in exact sequence."""
    app: ASGIApp = create_app()
    collector = SendEventCollector()

    scope: Scope = {
        "type": "http",
        "http_version": "1.1",
        "method": "GET",
        "path": "/ping",
        "raw_path": b"/ping",
        "query_string": b"",
        "headers": [(b"host", b"testserver")],
    }

    async def mock_receive() -> Message:
        return {"type": "http.request", "body": b"", "more_body": False}

    async def run_pipeline() -> None:
        await app(scope, mock_receive, collector.send)

    asyncio.run(run_pipeline())

    assert len(collector.messages) >= 2
    start_msg = collector.messages[0]
    assert start_msg["type"] == "http.response.start"
    assert start_msg["status"] == 200

    # Ensure header injector applied headers via the Send wrapper
    headers_dict = dict(start_msg["headers"])
    assert headers_dict[b"x-send-wrapper"] == b"active"

    body_msg = collector.messages[1]
    assert body_msg["type"] == "http.response.body"
    assert b"healthy" in body_msg["body"]


def test_direct_send_response_lifecycle_events() -> None:
    """Assertion 5: DirectSendResponse emits proper start and body events with more_body flags across multiple Send calls."""
    response = DirectSendResponse(chunks=[b"part1", b"part2", b"part3"], content_type="application/octet-stream")
    collector = SendEventCollector()

    scope: Scope = {
        "type": "http",
        "http_version": "1.1",
        "method": "GET",
        "path": "/stream",
        "raw_path": b"/stream",
        "query_string": b"",
        "headers": [],
    }

    async def mock_receive() -> Message:
        return {"type": "http.request", "body": b"", "more_body": False}

    async def run_response() -> None:
        await response(scope, mock_receive, collector.send)

    asyncio.run(run_response())

    assert len(collector.messages) == 4
    # Event 0: response.start
    assert collector.messages[0]["type"] == "http.response.start"
    assert collector.messages[0]["status"] == 200

    # Event 1: first body chunk (more_body=True)
    assert collector.messages[1]["type"] == "http.response.body"
    assert collector.messages[1]["body"] == b"part1"
    assert collector.messages[1]["more_body"] is True

    # Event 2: second body chunk (more_body=True)
    assert collector.messages[2]["type"] == "http.response.body"
    assert collector.messages[2]["body"] == b"part2"
    assert collector.messages[2]["more_body"] is True

    # Event 3: third body chunk (more_body=False)
    assert collector.messages[3]["type"] == "http.response.body"
    assert collector.messages[3]["body"] == b"part3"
    assert collector.messages[3]["more_body"] is False


def main() -> None:
    test_send_type_annotation_and_signature()
    test_send_wrapper_middleware_header_injection()
    test_chunked_streaming_via_send_callable()
    test_direct_asgi_send_lifecycle()
    test_direct_send_response_lifecycle_events()
    print("Contract verification succeeded for starlette.types.Send in starlette 1.6.0.")


if __name__ == "__main__":
    main()