Typed httpx client parsing into pydantic v2 models, tested with respx: working versions (tested 2026-10-15)

httpx==0.28.1
pydantic==2.14.1
respx==0.23.1
pytest==9.1.1

import httpx
from pydantic import BaseModel

class Repo(BaseModel):
    id: int
    full_name: str
    stargazers_count: int

class ApiClient:
    def __init__(self, base_url: str = "https://api.example.com", retries: int = 2):
        self.http = httpx.AsyncClient(base_url=base_url, timeout=10)
        self.retries = retries

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        await self.http.aclose()

    async def get_repo(self, owner: str, name: str) -> Repo:
        for attempt in range(self.retries + 1):
            try:
                r = await self.http.get(f"/repos/{owner}/{name}")
            except httpx.TransportError:
                if attempt == self.retries:
                    raise
                continue
            if r.status_code < 500 or attempt == self.retries:
                break
        r.raise_for_status()
        # validate raw bytes: invalid JSON also surfaces as pydantic ValidationError
        return Repo.model_validate_json(r.content)

import asyncio

import httpx
import pytest
import respx
from pydantic import ValidationError

from client import ApiClient, Repo

URL = "https://api.example.com/repos/a/b"
OK = {"id": 1, "full_name": "a/b", "stargazers_count": 5}


def run(coro):
    # sync tests + asyncio.run: no pytest-asyncio dependency needed
    return asyncio.run(coro)


async def fetch(owner="a", name="b"):
    async with ApiClient() as c:
        return await c.get_repo(owner, name)


@respx.mock
def test_parses_and_ignores_extra_fields():
    respx.get(URL).mock(return_value=httpx.Response(200, json={**OK, "extra": True}))
    assert run(fetch()) == Repo(id=1, full_name="a/b", stargazers_count=5)


@respx.mock
def test_retries_5xx_then_succeeds():
    route = respx.get(URL).mock(side_effect=[httpx.Response(503), httpx.Response(200, json=OK)])
    run(fetch())
    assert route.call_count == 2


@respx.mock
def test_retries_transport_errors():
    route = respx.get(URL).mock(side_effect=[httpx.ConnectTimeout("slow"), httpx.Response(200, json=OK)])
    assert run(fetch()).id == 1
    assert route.call_count == 2


@respx.mock
def test_4xx_is_not_retried():
    route = respx.get(URL).mock(return_value=httpx.Response(404))
    with pytest.raises(httpx.HTTPStatusError):
        run(fetch())
    assert route.call_count == 1


@respx.mock
def test_missing_field_raises_validation_error():
    respx.get(URL).mock(return_value=httpx.Response(200, json={"id": 1, "full_name": "a/b"}))
    with pytest.raises(ValidationError):
        run(fetch())


@respx.mock
def test_invalid_json_body_is_validation_error():
    respx.get(URL).mock(return_value=httpx.Response(200, content=b"<html>", headers={"content-type": "text/html"}))
    with pytest.raises(ValidationError):
        run(fetch())

[pytest]
pythonpath = .