Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion docs/decoders.md
Original file line number Diff line number Diff line change
Expand Up @@ -35,9 +35,13 @@ Both clients take `decoders: Sequence[ResponseDecoder] | None = None`. The list
- If `response_model=` is set and no decoder accepts it, the client raises `MissingDecoderError` before sending the request. That means nothing handles the type, and the fix is to install an extra or pass `decoders=[...]`. A `DecodeError` instead means a decoder ran and the body didn't fit the type, which points at the server or the model. See [Errors](errors.md).
- `MsgspecDecoder` accepts a `Struct`, dataclass, `TypedDict` or `NamedTuple` only if msgspec can decode every field type itself, nested fields included. A field typed as a pydantic model or some other class would need a `dec_hook`, which `MsgspecDecoder` doesn't take, so it declines the model and you get `MissingDecoderError` before the request goes out.

## Keep response models long-lived

The built-in decoders build a parser once per model (a `pydantic.TypeAdapter` or a `msgspec.json.Decoder`) and cache it, so define each response model once, at module level, and reuse it. Each decoder caches up to 4,096 models. When the cache is full, the next new model empties it. A model created per request, for example with `pydantic.create_model` or `msgspec.defstruct`, takes a new slot every time, so the cache keeps filling up and emptying, and every model seen after it empties gets its parser built again.

## One sync protocol for both clients

Middleware comes in sync and async versions, but there is only one `ResponseDecoder` protocol, used by both clients. `decode` is synchronous because the client has the whole body by the time it runs, so there is nothing to await. If a middleware returns a streamed response it never read, the client reads it before decoding. In `AsyncClient` that stream also has to support sync iteration, as `httpx2.ByteStream` does; otherwise the middleware should `await response.aread()` before returning it. The same decoder works with either client.
Middleware comes in sync and async versions, but there is only one `ResponseDecoder` protocol, used by both clients. `decode` is synchronous because the client has the whole body by the time it runs, so there is nothing to await. If a middleware returns a streamed response it never read, the client reads it before decoding. The stream has to match the client: async in `AsyncClient` and sync in `Client`, or the call raises `TypeError` (see [Middleware](middleware.md)). An in-memory body such as `content=b"..."` works in both. The same decoder works with either client.

## Writing your own

Expand Down
19 changes: 16 additions & 3 deletions src/httpware/decoders/_caching.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@

V = typing.TypeVar("V")

_MAX_ENTRIES = 4096

_Cache: typing.TypeAlias = dict[Hashable, tuple[object, V | None]]


Expand All @@ -28,14 +30,15 @@ def _lookup(cache: "_Cache[V]", model: object, build: Callable[[typing.Any], V |
"""Return the cached `build(model)`, building it on a miss; None means `model` cannot be decoded.

`build` must not raise. Concurrent first calls may build twice; the last write wins.
A full cache is emptied before the next insert, so a model seen again after that is rebuilt.
"""
try:
key = _key(model)
entry = cache.get(key)
except Exception: # noqa: BLE001 — unhashable models fall back to a repr key
return _lookup_by_repr(cache, model, build)
if entry is None:
entry = cache[key] = (model, build(model))
entry = _store(cache, key, model, build)
return entry[1]


Expand All @@ -48,5 +51,15 @@ def _lookup_by_repr(cache: "_Cache[V]", model: object, build: Callable[[typing.A
return entry[1]
except Exception: # noqa: BLE001 — a model that cannot be keyed is built uncached
return build(model)
entry = cache[key] = (model, build(model))
return entry[1]
return _store(cache, key, model, build)[1]


def _store(
cache: "_Cache[V]", key: Hashable, model: object, build: Callable[[typing.Any], V | None]
) -> tuple[object, V | None]:
"""Build and cache the entry for `model`, emptying `cache` first once it holds `_MAX_ENTRIES`."""
entry = (model, build(model))
if len(cache) >= _MAX_ENTRIES:
cache.clear()
cache[key] = entry
return entry
77 changes: 77 additions & 0 deletions tests/test_decoders_caching.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
"""The decoder adapter caches stay bounded when models are created per request."""

import concurrent.futures
import gc
import typing
import weakref

import msgspec
import pydantic
import pytest

from httpware.decoders import _caching
from httpware.decoders.msgspec import MsgspecDecoder
from httpware.decoders.pydantic import PydanticDecoder


_CAP = 8


@pytest.fixture(autouse=True)
def _small_cap(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(_caching, "_MAX_ENTRIES", _CAP)


def _decode_through_fresh_models(decoder: PydanticDecoder | MsgspecDecoder, models: typing.Iterator[type]) -> list:
refs = []
for model in models:
decoder.decode(b'{"x": 1}', model)
refs.append(weakref.ref(model))
gc.collect()
return refs


def test_pydantic_cache_is_bounded_and_releases_old_models() -> None:
decoder = PydanticDecoder()
models = (pydantic.create_model(f"M{i}", x=(int, ...)) for i in range(_CAP + 1))
refs = _decode_through_fresh_models(decoder, models)
assert len(decoder._adapters) <= _CAP # noqa: SLF001
assert refs[0]() is None


def test_msgspec_cache_is_bounded_and_drops_old_models() -> None:
"""Checks our cache only: older msgspec inspects through `typing.Tuple[...]`, whose LRU keeps recent models."""
decoder = MsgspecDecoder()
models = [msgspec.defstruct(f"S{i}", [("x", int)]) for i in range(_CAP + 1)]
for model in models:
decoder.decode(b'{"x": 1}', model)
assert len(decoder._adapters) <= _CAP # noqa: SLF001
assert all(cached is not models[0] for cached, _ in decoder._adapters.values()) # noqa: SLF001


def test_repr_keyed_cache_is_bounded() -> None:
decoder = PydanticDecoder()
for i in range(_CAP + 1):
decoder.decode(b"1", typing.Annotated[int, [i]]) # ty: ignore[invalid-argument-type]
assert len(decoder._adapters) <= _CAP # noqa: SLF001


def test_model_decodes_after_the_cache_is_cleared() -> None:
decoder = PydanticDecoder()
for i in range(_CAP + 1):
decoder.decode(b'{"x": 1}', pydantic.create_model(f"M{i}", x=(int, ...)))
model = pydantic.create_model("Late", x=(int, ...))
assert decoder.decode(b'{"x": 2}', model).model_dump() == {"x": 2}
assert decoder.can_decode(model)


@pytest.mark.stress
def test_clear_racing_with_lookups_returns_the_right_adapter() -> None:
models = [type(f"T{i}", (), {}) for i in range(_CAP * 4)]
cache: _caching._Cache[type] = {}

def look_up_all(_: int) -> bool:
return all(_caching._lookup(cache, model, lambda m: m) is model for _ in range(50) for model in models) # noqa: SLF001

with concurrent.futures.ThreadPoolExecutor(max_workers=16) as pool:
assert all(pool.map(look_up_all, range(32)))
Loading