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

Use EXACTLY these public packages and versions:
  - pkg:pypi/requests@2.34.2
Demonstrate these symbols/APIs:
  - 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:a817cd5f63943fe862ee2d0e2b479a52e8017360b6dd5790cec4db1814b68227","contract":["requests.Request constructs an un-prepared HTTP request with method, url, headers, params, and json payload","requests.Request prepare compiles request into PreparedRequest with query string and headers","Session prepare_request merges session headers and cookies with requests.Request instance","Session send dispatches PreparedRequest compiled from requests.Request and returns response"],"goal":"verify requests.Request in pkg:pypi/requests@2.34.2","kind":"HOW","packages":["pkg:pypi/requests@2.34.2"],"schemaVersion":1,"symbols":["requests.Request"]},"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/requests@2.34.2"],"schemaVersion":1,"subject":"pkg:pypi/requests@2.34.2","symbols":["requests.Request"],"verifierAdapter":"python@1"}

requests==2.34.2

"""Demonstration of requests.Request preparation and execution."""

from typing import Any, Dict, Optional
import requests


def build_request(
    method: str,
    url: str,
    headers: Optional[Dict[str, str]] = None,
    params: Optional[Dict[str, Any]] = None,
    data: Any = None,
    json: Any = None,
    auth: Optional[Any] = None,
    cookies: Optional[Dict[str, str]] = None,
) -> requests.Request:
    """Construct an un-prepared Request object."""
    return requests.Request(
        method=method,
        url=url,
        headers=headers,
        params=params,
        data=data,
        json=json,
        auth=auth,
        cookies=cookies,
    )


def prepare_request_standalone(req: requests.Request) -> requests.PreparedRequest:
    """Prepare a Request independently without a session context."""
    return req.prepare()


def prepare_request_in_session(
    session: requests.Session, req: requests.Request
) -> requests.PreparedRequest:
    """Prepare a Request within a session, inheriting session headers and cookies."""
    return session.prepare_request(req)


def execute_prepared_request(
    session: requests.Session,
    prepared: requests.PreparedRequest,
    timeout: float = 10.0,
) -> requests.Response:
    """Send a PreparedRequest through a Session and return the Response."""
    return session.send(prepared, timeout=timeout)

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

"""Contract verification test for requests.Request."""

import http.server
import json
import os
import sys
import threading

# 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__))))

import requests
from requests import Request, Session
from sample import (
    build_request,
    prepare_request_standalone,
    prepare_request_in_session,
    execute_prepared_request,
)


class MockHTTPHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path.startswith("/api/resource"):
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("X-Echo-Custom", self.headers.get("X-Custom-Header", ""))
            self.send_header("X-Echo-Session", self.headers.get("X-Session-Header", ""))
            self.end_headers()
            self.wfile.write(b'{"status": "ok", "message": "retrieved"}')
        else:
            self.send_response(404)
            self.end_headers()

    def do_POST(self):
        if self.path.startswith("/api/submit"):
            content_length = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(content_length)
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("X-Echo-Custom", self.headers.get("X-Custom-Header", ""))
            self.end_headers()
            response_data = {
                "status": "created",
                "received_length": content_length,
                "received_data": json.loads(body.decode("utf-8")) if body else {},
            }
            self.wfile.write(json.dumps(response_data).encode("utf-8"))
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, format, *args):
        # Suppress server access logs during tests
        pass


def test_request_initialization() -> None:
    """Assertion 1: requests.Request constructs an un-prepared HTTP request with method, url, headers, params, and json payload."""
    req = build_request(
        method="POST",
        url="http://127.0.0.1:8080/api/submit",
        headers={"X-Custom-Header": "header-val-1"},
        params={"category": "test", "page": "1"},
        json={"item": "widget", "qty": 5},
    )

    assert isinstance(req, Request)
    assert req.method == "POST"
    assert req.url == "http://127.0.0.1:8080/api/submit"
    assert req.headers == {"X-Custom-Header": "header-val-1"}
    assert req.params == {"category": "test", "page": "1"}
    assert req.json == {"item": "widget", "qty": 5}


def test_request_standalone_preparation() -> None:
    """Assertion 2: requests.Request prepare compiles request into PreparedRequest with query string and headers."""
    req = build_request(
        method="POST",
        url="http://127.0.0.1:8080/api/submit",
        headers={"X-Custom-Header": "header-val-1"},
        params={"category": "test", "page": "1"},
        json={"item": "widget", "qty": 5},
    )

    prepared = prepare_request_standalone(req)
    assert isinstance(prepared, requests.PreparedRequest)
    assert prepared.method == "POST"
    assert prepared.url == "http://127.0.0.1:8080/api/submit?category=test&page=1"
    assert prepared.headers.get("X-Custom-Header") == "header-val-1"
    assert prepared.headers.get("Content-Type") == "application/json"
    assert prepared.body is not None
    body_json = json.loads(prepared.body.decode("utf-8"))
    assert body_json == {"item": "widget", "qty": 5}


def test_request_session_preparation() -> None:
    """Assertion 3: Session prepare_request merges session headers and cookies with requests.Request instance."""
    session = Session()
    session.headers["X-Session-Header"] = "session-token-42"
    session.cookies.set("session_cookie", "cookie_value_99")

    req = build_request(
        method="GET",
        url="http://127.0.0.1:8080/api/resource",
        headers={"X-Custom-Header": "request-level-header"},
    )

    prepared = prepare_request_in_session(session, req)
    assert isinstance(prepared, requests.PreparedRequest)
    assert prepared.headers.get("X-Session-Header") == "session-token-42"
    assert prepared.headers.get("X-Custom-Header") == "request-level-header"
    assert "session_cookie=cookie_value_99" in prepared.headers.get("Cookie", "")


def test_request_execution(base_url: str) -> None:
    """Assertion 4: Session send dispatches PreparedRequest compiled from requests.Request and returns response."""
    session = Session()
    session.headers["X-Session-Header"] = "active-session"

    # Test POST prepared request
    post_req = build_request(
        method="POST",
        url=f"{base_url}/api/submit",
        headers={"X-Custom-Header": "post-header"},
        json={"name": "test-entry", "count": 10},
    )
    post_prep = prepare_request_in_session(session, post_req)
    post_resp = execute_prepared_request(session, post_prep, timeout=5.0)

    assert post_resp.status_code == 200
    assert post_resp.headers.get("X-Echo-Custom") == "post-header"
    post_data = post_resp.json()
    assert post_data["status"] == "created"
    assert post_data["received_data"] == {"name": "test-entry", "count": 10}

    # Test GET prepared request
    get_req = build_request(
        method="GET",
        url=f"{base_url}/api/resource",
        headers={"X-Custom-Header": "get-header"},
    )
    get_prep = prepare_request_in_session(session, get_req)
    get_resp = execute_prepared_request(session, get_prep, timeout=5.0)

    assert get_resp.status_code == 200
    assert get_resp.headers.get("X-Echo-Custom") == "get-header"
    assert get_resp.headers.get("X-Echo-Session") == "active-session"
    assert get_resp.json() == {"status": "ok", "message": "retrieved"}


def main() -> None:
    server = http.server.HTTPServer(("127.0.0.1", 0), MockHTTPHandler)
    port = server.server_address[1]
    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
    server_thread.start()

    base_url = f"http://127.0.0.1:{port}"
    try:
        test_request_initialization()
        test_request_standalone_preparation()
        test_request_session_preparation()
        test_request_execution(base_url)
        print("All contract assertions passed.")
    finally:
        server.shutdown()
        server.server_close()


if __name__ == "__main__":
    main()