httpx 0.28.1: Register synchronous and asynchronous request and response event hooks in httpx Client and AsyncClient to transform headers, validate status codes, and intercept multi-hop redirects

# Notes: httpx Event Hooks Lifecycle and Interceptors (0.28.1)

## 1. What `search_known_solution` Answered
`search_known_solution` returned existing samples covering:
- Timeout defaults (`httpx.Client()` defaults to 5.0s across phases)
- Stateful `httpx.Auth` generators and Basic/Digest auth
- RFC 3986 URL join semantics vs `Client(base_url=...)`
- Redirect handling (`follow_redirects=True` vs requests `allow_redirects`)
- Error hierarchy (`httpx.InvalidURL` directly inheriting from `Exception`)
- Headers and QueryParams multi-value dictionary conversion and immutability

None of the existing samples verified request and response event hooks lifecycle, exception propagation in hooks, intermediate redirect hook execution, or asynchronous hook support in `httpx.AsyncClient`.

## 2. Exact Release Pinned
- **Package**: `httpx`
- **Pinned Version**: `0.28.1`
- **Manifest**: `requirements.txt`
- **Lockfile**: `requirements.lock`

## 3. Observed Behaviour
1. **Synchronous Request and Response Event Hooks**: Registered at client instantiation via `event_hooks={"request": [...], "response": [...]}`. Request hooks receive `(request,)` before dispatch and can mutate headers or record outbound metrics. Response hooks receive `(response,)` after receipt.
2. **Status Validation and Exception Propagation in Hooks**: A response hook calling `response.raise_for_status()` turns 4xx and 5xx responses into `httpx.HTTPStatusError` exceptions before control returns to the caller. Exceptions raised within hooks propagate cleanly.
3. **Multi-Hop Redirect Hook Lifecycle**: When `follow_redirects=True`, response event hooks are triggered for every intermediate redirect response (e.g. 302 Foundation) in sequence, as well as the final destination response.
4. **Asynchronous Hook Execution in `AsyncClient`**: `httpx.AsyncClient` natively awaits coroutines supplied in `event_hooks`, allowing non-blocking async transformations and logging.
5. **Client-Level Configuration Strictness**: `event_hooks` is strictly a client configuration argument. Passing `event_hooks` to individual request methods (`client.get()`, `client.request()`) raises `TypeError`.
6. **Deterministic Offline Verification**: `httpx.MockTransport` enables complete offline contract testing of hook execution order and state changes without real network I/O.

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: Register synchronous and asynchronous request and response event hooks in httpx Client and AsyncClient to transform headers, validate status codes, and intercept multi-hop redirects
Kind: HOW

Use EXACTLY these public packages and versions:
  - pkg:pypi/httpx@0.28.1
Demonstrate these symbols/APIs:
  - httpx.AsyncClient
  - httpx.Client
  - httpx.HTTPStatusError
  - httpx.MockTransport
  - httpx.Request
  - httpx.Response

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:b1d81ba9bdd20d252a60c3e9edbe5f114211a168239b3f95eef8d704554fb5ce","contract":["httpx.Client executes request event hooks sequentially before dispatch, allowing synchronous header mutations.","httpx.Client executes response event hooks after response receipt, enabling status checks like response.raise_for_status() to propagate HTTPStatusError.","httpx.Client with follow_redirects=True invokes response event hooks for every intermediate redirect response and the final response.","httpx.Client raises TypeError when event_hooks is passed per-request rather than configured on the client instance.","httpx.AsyncClient executes and awaits asynchronous coroutine event hooks across request and response phases.","httpx.MockTransport enables offline verification of request and response event hooks lifecycle without network access."],"goal":"Register synchronous and asynchronous request and response event hooks in httpx Client and AsyncClient to transform headers, validate status codes, and intercept multi-hop redirects","kind":"HOW","packages":["pkg:pypi/httpx@0.28.1"],"schemaVersion":1,"symbols":["httpx.AsyncClient","httpx.Client","httpx.HTTPStatusError","httpx.MockTransport","httpx.Request","httpx.Response"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","distro":"ubuntu","ecosystem":"pypi","executionContext":"python","language":"python","libc":"glibc","libcVersion":"2.39","os":"linux","osVersionBucket":"24","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/httpx@0.28.1"],"schemaVersion":1,"subject":"pkg:pypi/httpx@0.28.1","symbols":["httpx.AsyncClient","httpx.Client","httpx.HTTPStatusError","httpx.MockTransport","httpx.Request","httpx.Response"],"verifierAdapter":"python@1"}

anyio==4.14.2
certifi==2026.7.22
h11==0.16.0
httpcore==1.0.9
httpx==0.28.1
idna==3.19
sniffio==1.3.1
typing-extensions==4.16.0

anyio==4.14.2
certifi==2026.7.22
h11==0.16.0
httpcore==1.0.9
httpx==0.28.1
idna==3.19
sniffio==1.3.1
typing-extensions==4.16.0

{
  "schemaVersion": 1,
  "goal": "Register synchronous and asynchronous request and response event hooks in httpx Client and AsyncClient to transform headers, validate status codes, and intercept multi-hop redirects",
  "kind": "HOW",
  "packages": [
    "pkg:pypi/httpx@0.28.1"
  ],
  "symbols": [
    "httpx.AsyncClient",
    "httpx.Client",
    "httpx.HTTPStatusError",
    "httpx.MockTransport",
    "httpx.Request",
    "httpx.Response"
  ]
}

"""Event hooks verification package for httpx."""

"""Event hook helpers and interceptors for httpx.Client and httpx.AsyncClient."""

from typing import Callable, Dict, List, Optional
import httpx


def add_correlation_id_hook(header_name: str = "X-Correlation-ID", correlation_id: str = "corr-default-123") -> Callable[[httpx.Request], None]:
    """Return a request event hook that attaches a correlation ID header."""
    def hook(request: httpx.Request) -> None:
        request.headers[header_name] = correlation_id
    return hook


def raise_for_status_hook(response: httpx.Response) -> None:
    """Response event hook that automatically invokes response.raise_for_status()."""
    response.raise_for_status()


class EventRecorder:
    """Collects chronological trace of request and response hook invocations."""

    def __init__(self) -> None:
        self.events: List[str] = []

    def request_hook(self, request: httpx.Request) -> None:
        self.events.append(f"request:{request.method}:{request.url.path}")

    def response_hook(self, response: httpx.Response) -> None:
        self.events.append(f"response:{response.status_code}:{response.request.url.path}")

    async def async_request_hook(self, request: httpx.Request) -> None:
        self.events.append(f"async_request:{request.method}:{request.url.path}")

    async def async_response_hook(self, response: httpx.Response) -> None:
        self.events.append(f"async_response:{response.status_code}:{response.request.url.path}")


def create_mock_client(
    handler: Callable[[httpx.Request], httpx.Response],
    event_hooks: Optional[Dict[str, List[Callable[..., None]]]] = None,
    follow_redirects: bool = False,
    base_url: str = "https://example.com",
) -> httpx.Client:
    """Create an offline httpx.Client with MockTransport and optional event_hooks."""
    transport = httpx.MockTransport(handler)
    return httpx.Client(
        transport=transport,
        event_hooks=event_hooks,
        follow_redirects=follow_redirects,
        base_url=base_url,
    )


def create_mock_async_client(
    handler: Callable[[httpx.Request], httpx.Response],
    event_hooks: Optional[Dict[str, List[Callable[..., None]]]] = None,
    follow_redirects: bool = False,
    base_url: str = "https://example.com",
) -> httpx.AsyncClient:
    """Create an offline httpx.AsyncClient with MockTransport and optional event_hooks."""
    transport = httpx.MockTransport(handler)
    return httpx.AsyncClient(
        transport=transport,
        event_hooks=event_hooks,
        follow_redirects=follow_redirects,
        base_url=base_url,
    )

"""Test package for event hooks contract verification."""

"""Contract verification for httpx request and response event hooks lifecycle."""

import asyncio
import sys
from pathlib import Path

# Ensure project root is in sys.path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import httpx
from src.hooks import (
    EventRecorder,
    add_correlation_id_hook,
    create_mock_async_client,
    create_mock_client,
    raise_for_status_hook,
)


def test_sync_event_hooks_lifecycle_and_header_mutation() -> None:
    """Prove request hooks mutate headers sequentially and response hooks process replies."""
    recorder = EventRecorder()
    custom_corr_id = "corr-test-xyz-987"

    def mock_handler(request: httpx.Request) -> httpx.Response:
        # Verify that request hook modified the header before transport handler received it
        received_corr = request.headers.get("X-Correlation-ID")
        assert received_corr == custom_corr_id, f"Expected {custom_corr_id}, got {received_corr}"
        return httpx.Response(status_code=200, json={"status": "ok", "corr_id": received_corr})

    client = create_mock_client(
        handler=mock_handler,
        event_hooks={
            "request": [add_correlation_id_hook("X-Correlation-ID", custom_corr_id), recorder.request_hook],
            "response": [recorder.response_hook],
        },
        base_url="https://example.com",
    )

    with client:
        response = client.get("/api/v1/resource")
        assert response.status_code == 200
        assert response.json()["corr_id"] == custom_corr_id
        assert recorder.events == [
            "request:GET:/api/v1/resource",
            "response:200:/api/v1/resource",
        ]


def test_response_hook_raise_for_status_propagates_http_status_error() -> None:
    """Prove response hook calling raise_for_status halts execution and raises HTTPStatusError."""
    recorder = EventRecorder()

    def mock_handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            status_code=404,
            json={"error": "resource_not_found"},
            request=request,
        )

    client = create_mock_client(
        handler=mock_handler,
        event_hooks={
            "response": [raise_for_status_hook, recorder.response_hook],
        },
        base_url="https://example.com",
    )

    with client:
        raised = False
        try:
            client.get("/api/v1/nonexistent")
        except httpx.HTTPStatusError as exc:
            raised = True
            assert exc.response.status_code == 404
            assert exc.response.json() == {"error": "resource_not_found"}

        assert raised, "Expected httpx.HTTPStatusError to be raised by response hook"
        # Since raise_for_status_hook raised first, subsequent response hooks in the list are aborted
        assert recorder.events == []


def test_response_hooks_fire_on_intermediate_redirects() -> None:
    """Prove response hooks execute for every intermediate redirect response and the final target."""
    seen_statuses = []

    def status_logger(response: httpx.Response) -> None:
        seen_statuses.append(response.status_code)

    def mock_handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/source":
            return httpx.Response(
                status_code=301,
                headers={"Location": "https://example.com/step2"},
                request=request,
            )
        elif request.url.path == "/step2":
            return httpx.Response(
                status_code=302,
                headers={"Location": "https://example.com/destination"},
                request=request,
            )
        elif request.url.path == "/destination":
            return httpx.Response(
                status_code=200,
                text="Arrived at final destination",
                request=request,
            )
        return httpx.Response(status_code=400, request=request)

    client = create_mock_client(
        handler=mock_handler,
        follow_redirects=True,
        event_hooks={"response": [status_logger]},
        base_url="https://example.com",
    )

    with client:
        resp = client.get("/source")
        assert resp.status_code == 200
        assert resp.text == "Arrived at final destination"
        # Two intermediate redirects captured in response.history
        assert len(resp.history) == 2
        assert resp.history[0].status_code == 301
        assert resp.history[1].status_code == 302
        # Response hook was invoked across all 3 responses in order
        assert seen_statuses == [301, 302, 200]


def test_event_hooks_cannot_be_passed_per_request() -> None:
    """Prove passing event_hooks to client.get/request raises TypeError."""
    def mock_handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(status_code=200, request=request)

    client = create_mock_client(handler=mock_handler, base_url="https://example.com")

    with client:
        type_error_raised = False
        try:
            client.get("/test", event_hooks={"request": []})  # type: ignore[call-arg]
        except TypeError as exc:
            type_error_raised = True
            assert "event_hooks" in str(exc)

        assert type_error_raised, "Expected TypeError when passing event_hooks per-request"


def test_async_client_event_hooks_await_coroutines() -> None:
    """Prove httpx.AsyncClient awaits async request and response event hooks."""
    recorder = EventRecorder()

    def mock_handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            status_code=200,
            json={"async_result": "success"},
            request=request,
        )

    async def run_async_test() -> None:
        client = create_mock_async_client(
            handler=mock_handler,
            event_hooks={
                "request": [recorder.async_request_hook],
                "response": [recorder.async_response_hook],
            },
            base_url="https://example.com",
        )
        async with client:
            resp = await client.get("/async/endpoint")
            assert resp.status_code == 200
            assert resp.json() == {"async_result": "success"}

    asyncio.run(run_async_test())
    assert recorder.events == [
        "async_request:GET:/async/endpoint",
        "async_response:200:/async/endpoint",
    ]


def main() -> None:
    test_sync_event_hooks_lifecycle_and_header_mutation()
    test_response_hook_raise_for_status_propagates_http_status_error()
    test_response_hooks_fire_on_intermediate_redirects()
    test_event_hooks_cannot_be_passed_per_request()
    test_async_client_event_hooks_await_coroutines()
    print("Contract pass: httpx request and response event hooks lifecycle verified cleanly.")


if __name__ == "__main__":
    main()