starlette 1.6.0: types.Receive

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

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:eb59b072b44dcd327458946b99d7c7d7de995076a96b12e8c1125ddb42c6464d","contract":["assert Receive callable consumes streamed HTTP request body chunks until more_body is False","assert Receive callable cleanly handles http.disconnect message during body consumption","assert Starlette Request resolves complete body when initialized with custom Receive callable","assert BufferedReceiveMiddleware wraps Receive callable and forwards ASGI messages downstream","assert starlette.types.Receive type annotations match ASGI 3.0 specification"],"goal":"verify starlette.types.Receive in pkg:pypi/starlette@1.6.0","kind":"HOW","packages":["pkg:pypi/starlette@1.6.0"],"schemaVersion":1,"symbols":["starlette.types.Receive"]},"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.Receive"],"verifierAdapter":"python@1"}

[project]
name = "verify-starlette-receive"
version = "0.1.0"
dependencies = [
    "starlette==1.6.0",
    "anyio>=3.4.0",
    "typing-extensions>=4.0.0",
]

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

"""Sample demonstrating custom ASGI body consumption and middleware using starlette.types.Receive."""

from typing import List
from starlette.types import Receive, Scope, Send, Message
from starlette.responses import JSONResponse


class StreamBodyReceiver:
    """Consumes incoming request body chunks from an ASGI Receive callable."""

    def __init__(self, receive: Receive) -> None:
        self.receive = receive

    async def read_full_body(self) -> bytes:
        chunks: List[bytes] = []
        while True:
            message: Message = await self.receive()
            msg_type = message.get("type")
            if msg_type == "http.request":
                chunk = message.get("body", b"")
                if chunk:
                    chunks.append(chunk)
                if not message.get("more_body", False):
                    break
            elif msg_type == "http.disconnect":
                break
        return b"".join(chunks)


class BufferedReceiveMiddleware:
    """ASGI middleware that wraps Receive to inspect and pass along request bodies."""

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

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

        received_log: List[str] = []

        async def buffered_receive() -> Message:
            msg = await receive()
            if msg.get("type") == "http.request":
                body_len = len(msg.get("body", b""))
                received_log.append(f"chunk:{body_len}")
            elif msg.get("type") == "http.disconnect":
                received_log.append("disconnect")
            return msg

        scope["app.buffered_log"] = received_log
        await self.app(scope, buffered_receive, send)


async def raw_asgi_app(scope: Scope, receive: Receive, send: Send) -> None:
    """Raw ASGI application directly consuming Receive callable."""
    if scope.get("type") != "http":
        return

    receiver = StreamBodyReceiver(receive)
    body = await receiver.read_full_body()

    response = JSONResponse(
        {
            "received_bytes": len(body),
            "content": body.decode("utf-8") if body else "",
            "log": scope.get("app.buffered_log", []),
        }
    )
    await response(scope, receive, send)

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

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

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

# 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 starlette.types import Receive, Scope, Send, Message
from sample import StreamBodyReceiver, BufferedReceiveMiddleware, raw_asgi_app


def test_receive_streamed_chunks() -> None:
    """Assertion 1: assert Receive callable consumes streamed HTTP request body chunks until more_body is False"""
    chunk_list = [b"part1-", b"part2-", b"part3"]
    chunk_index = 0

    async def mock_receive() -> Message:
        nonlocal chunk_index
        if chunk_index < len(chunk_list):
            chunk = chunk_list[chunk_index]
            chunk_index += 1
            is_last = chunk_index >= len(chunk_list)
            return {"type": "http.request", "body": chunk, "more_body": not is_last}
        return {"type": "http.request", "body": b"", "more_body": False}

    async def run() -> None:
        receiver = StreamBodyReceiver(mock_receive)
        full_body = await receiver.read_full_body()
        assert full_body == b"part1-part2-part3", f"Unexpected body: {full_body}"
        assert chunk_index == 3, f"Expected 3 chunks read, got {chunk_index}"

    asyncio.run(run())


def test_receive_disconnect_handling() -> None:
    """Assertion 2: assert Receive callable cleanly handles http.disconnect message during body consumption"""
    calls = 0

    async def disconnect_receive() -> Message:
        nonlocal calls
        calls += 1
        if calls == 1:
            return {"type": "http.request", "body": b"initial_chunk", "more_body": True}
        return {"type": "http.disconnect"}

    async def run() -> None:
        receiver = StreamBodyReceiver(disconnect_receive)
        result = await receiver.read_full_body()
        assert result == b"initial_chunk", f"Unexpected body on disconnect: {result}"
        assert calls == 2

    asyncio.run(run())


def test_starlette_request_with_custom_receive() -> None:
    """Assertion 3: assert Starlette Request resolves complete body when initialized with custom Receive callable"""
    scope: Scope = {
        "type": "http",
        "http_version": "1.1",
        "method": "POST",
        "path": "/submit",
        "raw_path": b"/submit",
        "query_string": b"",
        "headers": [(b"content-type", b"text/plain")],
    }

    message_queue = [
        {"type": "http.request", "body": b"hello ", "more_body": True},
        {"type": "http.request", "body": b"world", "more_body": False},
    ]

    async def custom_receive() -> Message:
        if message_queue:
            return message_queue.pop(0)
        return {"type": "http.request", "body": b"", "more_body": False}

    async def run() -> None:
        req = Request(scope, receive=custom_receive)
        body = await req.body()
        assert body == b"hello world", f"Unexpected request body: {body}"

    asyncio.run(run())


def test_buffered_receive_middleware() -> None:
    """Assertion 4: assert BufferedReceiveMiddleware wraps Receive callable and forwards ASGI messages downstream"""
    scope: Scope = {
        "type": "http",
        "http_version": "1.1",
        "method": "POST",
        "path": "/test",
        "raw_path": b"/test",
        "query_string": b"",
        "headers": [],
    }

    sent_messages: List[Message] = []

    async def mock_send(msg: Message) -> None:
        sent_messages.append(msg)

    raw_chunks = [
        {"type": "http.request", "body": b"payload-chunk-1", "more_body": True},
        {"type": "http.request", "body": b"-and-chunk-2", "more_body": False},
    ]

    async def original_receive() -> Message:
        if raw_chunks:
            return raw_chunks.pop(0)
        return {"type": "http.request", "body": b"", "more_body": False}

    async def run() -> None:
        app = BufferedReceiveMiddleware(raw_asgi_app)
        await app(scope, original_receive, mock_send)

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

        assert status_code == 200, f"Expected 200, got {status_code}"
        data = json.loads(response_body.decode("utf-8"))
        assert data["received_bytes"] == len(b"payload-chunk-1-and-chunk-2")
        assert data["content"] == "payload-chunk-1-and-chunk-2"
        assert data["log"] == ["chunk:15", "chunk:12"]

    asyncio.run(run())


def test_receive_type_annotation_contract() -> None:
    """Assertion 5: assert starlette.types.Receive type annotations match ASGI 3.0 specification"""
    async def valid_receive_impl() -> Message:
        return {"type": "http.request", "body": b"", "more_body": False}

    def accepts_receive_target(fn: Receive) -> bool:
        return callable(fn)

    assert accepts_receive_target(valid_receive_impl) is True
    assert Receive is not None


def main() -> None:
    test_receive_streamed_chunks()
    test_receive_disconnect_handling()
    test_starlette_request_with_custom_receive()
    test_buffered_receive_middleware()
    test_receive_type_annotation_contract()
    print("Starlette Receive contract verification passed.")


if __name__ == "__main__":
    main()