urllib3 2.7.0: Retry, Timeout

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 pkg:pypi/urllib3@2.7.0
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/urllib3@2.7.0

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:f4981e8ea7d402f2c7a7c9dbd129842eaeb14df7dc2d991d6452979e7dca6d68","contract":["urllib3.Retry configures retry attempts and status forcelist for automatic request retries","urllib3.Retry calculates backoff delays and manages remaining retry counts","urllib3.Timeout configures granular connect and read timeout limits on HTTP requests"],"goal":"verify pkg:pypi/urllib3@2.7.0","kind":"HOW","packages":["pkg:pypi/urllib3@2.7.0"],"schemaVersion":1,"symbols":["urllib3.Retry","urllib3.Timeout"]},"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/urllib3@2.7.0"],"schemaVersion":1,"symbols":["urllib3.Retry","urllib3.Timeout"],"verifierAdapter":"python@1"}

urllib3==2.7.0

"""Sample demonstrating request retries and timeout configuration using urllib3."""

from typing import Any, Dict, Optional, Tuple, Union
import urllib3
from urllib3.response import BaseHTTPResponse
from urllib3.util import Retry, Timeout


def create_retry_policy(
    total: int = 3,
    backoff_factor: float = 0.1,
    status_forcelist: Tuple[int, ...] = (500, 502, 503, 504),
    raise_on_status: bool = False,
) -> Retry:
    """Create a configured Retry policy for handling transient errors."""
    return Retry(
        total=total,
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
        raise_on_status=raise_on_status,
    )


def create_timeout_config(
    connect: float = 1.0,
    read: float = 2.0,
) -> Timeout:
    """Create a granular connect and read Timeout configuration."""
    return Timeout(connect=connect, read=read)


def create_resilient_pool(
    retries: Optional[Union[Retry, int]] = None,
    timeout: Optional[Union[Timeout, float]] = None,
    num_pools: int = 5,
    maxsize: int = 5,
) -> urllib3.PoolManager:
    """Create a urllib3 PoolManager with custom retry policy and timeout settings."""
    return urllib3.PoolManager(
        retries=retries,
        timeout=timeout,
        num_pools=num_pools,
        maxsize=maxsize,
    )


def send_request_with_policy(
    http: urllib3.PoolManager,
    method: str,
    url: str,
    retries: Optional[Union[Retry, int, bool]] = None,
    timeout: Optional[Union[Timeout, float]] = None,
    headers: Optional[Dict[str, str]] = None,
) -> BaseHTTPResponse:
    """Send an HTTP request using the pool with optional per-request retries and timeout override."""
    kwargs: Dict[str, Any] = {}
    if retries is not None:
        kwargs["retries"] = retries
    if timeout is not None:
        kwargs["timeout"] = timeout
    if headers is not None:
        kwargs["headers"] = headers
    return http.request(method, url, **kwargs)


def get_next_retry(
    retry: Retry,
    method: str = "GET",
    url: str = "http://localhost/",
    response_status: int = 503,
) -> Retry:
    """Simulate a retry decrement for a given HTTP status code."""
    return retry.increment(method=method, url=url, response=None)

{
  "schemaVersion": 1,
  "goal": "verify pkg:pypi/urllib3@2.7.0",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/urllib3@2.7.0"
  ]
}

"""Contract verification test for urllib3 Retry and Timeout policies."""

import os
import sys
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

# 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 urllib3
from urllib3.util import Retry, Timeout
from sample import (
    create_resilient_pool,
    create_retry_policy,
    create_timeout_config,
    get_next_retry,
    send_request_with_policy,
)


class RetryMockServerHandler(BaseHTTPRequestHandler):
    """Mock HTTP handler that fails a configured number of times before succeeding."""

    request_count = 0

    def do_GET(self):
        RetryMockServerHandler.request_count += 1
        if self.path == "/transient-endpoint":
            if RetryMockServerHandler.request_count < 3:
                self.send_response(503)
                self.send_header("Content-Type", "text/plain")
                self.end_headers()
                self.wfile.write(b"Service Unavailable - retry later")
            else:
                self.send_response(200)
                self.send_header("Content-Type", "text/plain")
                self.end_headers()
                self.wfile.write(b"Success after retries")
        elif self.path == "/health":
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.end_headers()
            self.wfile.write(b"OK")
        else:
            self.send_response(404)
            self.end_headers()

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


def run_mock_server():
    RetryMockServerHandler.request_count = 0
    server = HTTPServer(("127.0.0.1", 0), RetryMockServerHandler)
    host, port = server.server_address
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, f"http://127.0.0.1:{port}"


def test_retry_on_status_forcelist(base_url: str) -> None:
    """Assertion 1: urllib3.Retry configures retry attempts and status forcelist for automatic request retries."""
    retry_policy = create_retry_policy(
        total=3,
        backoff_factor=0.01,
        status_forcelist=(503,),
        raise_on_status=False,
    )
    assert retry_policy.total == 3, "Retry total should be 3"
    assert 503 in retry_policy.status_forcelist, "503 should be in status_forcelist"

    http = create_resilient_pool(retries=retry_policy)
    target_url = f"{base_url}/transient-endpoint"

    response = send_request_with_policy(http, "GET", target_url)
    assert response.status == 200, f"Expected status 200 after retries, got {response.status}"
    assert response.data == b"Success after retries", "Response body mismatch"
    assert RetryMockServerHandler.request_count == 3, (
        f"Expected exactly 3 server hits, got {RetryMockServerHandler.request_count}"
    )


def test_retry_backoff_and_decrement() -> None:
    """Assertion 2: urllib3.Retry calculates backoff delays and manages remaining retry counts."""
    initial_retry = Retry(
        total=4,
        backoff_factor=1.0,
        status_forcelist=(500, 502, 503),
    )
    assert initial_retry.total == 4, "Initial retry count should be 4"

    # Simulate first failure
    first_step = get_next_retry(initial_retry, method="GET", url="http://127.0.0.1/api", response_status=503)
    assert first_step.total == 3, f"Expected 3 retries remaining, got {first_step.total}"
    assert len(first_step.history) == 1, "History length should be 1"

    # Simulate second failure
    second_step = get_next_retry(first_step, method="GET", url="http://127.0.0.1/api", response_status=503)
    assert second_step.total == 2, f"Expected 2 retries remaining, got {second_step.total}"
    assert len(second_step.history) == 2, "History length should be 2"

    # Backoff time calculation
    backoff_time = second_step.get_backoff_time()
    assert backoff_time > 0, f"Expected positive backoff time, got {backoff_time}"


def test_timeout_configuration(base_url: str) -> None:
    """Assertion 3: urllib3.Timeout configures granular connect and read timeout limits on HTTP requests."""
    timeout_config = create_timeout_config(connect=2.5, read=5.0)
    assert isinstance(timeout_config, Timeout), "Should be an instance of Timeout"
    assert timeout_config.connect_timeout == 2.5, "Connect timeout mismatch"
    assert timeout_config.read_timeout == 5.0, "Read timeout mismatch"

    http = create_resilient_pool(timeout=timeout_config)
    response = send_request_with_policy(http, "GET", f"{base_url}/health")
    assert response.status == 200, f"Expected status 200, got {response.status}"
    assert response.data == b"OK", "Health response mismatch"


def main() -> None:
    server, base_url = run_mock_server()
    try:
        test_retry_on_status_forcelist(base_url)
        test_retry_backoff_and_decrement()
        test_timeout_configuration(base_url)
        print("All contract assertions passed successfully.")
    finally:
        server.shutdown()
        server.server_close()


if __name__ == "__main__":
    main()