Caching¶

OxideHTTP supports asynchronous response caching through a pluggable cache backend.

Two implementations are included:

• MemoryCache — process-local in-memory cache.
• RedisCache — shared cache backed by Redis.

Both implement CacheProtocol, so you can also provide your own backend.

Cache backends¶

MemoryCache¶

MemoryCache stores responses in the current process.

from oxidehttp.cache.memory import MemoryCache

cache = MemoryCache()

Characteristics:

• Uses an asynchronous lock to protect cache entries.
• Uses a monotonic clock for expiration.
• Removes expired entries when they are accessed.
• Stores response status, headers, and body.
• Does not share entries between processes.
• Does not impose a maximum number of entries.

You can inspect the number of stored entries with len():

entry_count = await cache.__len__()

RedisCache¶

RedisCache stores serialized responses in Redis, allowing multiple application instances to share cached entries.

Install the Redis dependency:

uv add oxidehttp[redis] --index OxideHTTP

Create a cache backend using an asynchronous Redis client:

from redis.asyncio import Redis

from oxidehttp.cache.redis import RedisCache

redis = Redis.from_url('redis://localhost:6379/0')
cache = RedisCache(redis)

The default key prefix is:

oxidehttp:cache:

You can customize it:

cache = RedisCache(redis, prefix='myapp:cache:')

Redis entries use the supplied TTL, converted to milliseconds. Response bodies are encoded as hexadecimal strings in the serialized payload.

Remember to close your Redis client when your application shuts down:

await redis.aclose()

Configuring the client¶

Pass a cache backend to OxideHTTP through the cache argument:

from oxidehttp import OxideHTTP
from oxidehttp.cache.memory import MemoryCache

cache = MemoryCache()

async with OxideHTTP(cache=cache) as client:
    ...

The client accepts any implementation satisfying CacheProtocol.

Cache expiration¶

Cache lifetimes are specified in seconds.

A positive cache_ttl enables cache lookup. When a cached response exists and has not expired, the client returns it without sending another HTTP request.

response = await client.get('https://api.example.com/users', cache_ttl=60)

A TTL less than or equal to zero disables cache lookup and storage.

If no cache backend is configured while a positive TTL is supplied, the request proceeds normally and a warning is logged.

Cache keys¶

OxideHTTP generates a SHA-256 cache key from the following values:

• Request URL
• Query parameters
• Explicit cache_key value

The generated key does not include request headers or request body contents.

This means requests with the same URL, parameters, and explicit cache key can share a cache entry even if their headers or bodies differ.

An explicit cache key is also required when storing a response. The cache lookup can happen without one, but the client only writes a response when all of these conditions are met:

• An explicit cache_key is provided.
• A cache backend is configured.
• cache_ttl is positive.

API note: The internal _request() method accepts cache_key, but the public get() method currently exposes only cache_ttl. If you intend to use response caching through the public API, consider exposing cache_key there as well.

Cache management¶

Both built-in backends implement the following operations:

For example:

await cache.delete('some-cache-key')

await cache.clear()

RedisCache.clear() removes keys matching its configured prefix. Use a dedicated prefix for each application or cache namespace to avoid deleting unrelated entries.

Custom cache backends¶

You can implement your own backend by satisfying CacheProtocol:

from typing import Protocol

from oxidehttp.schema import CachedResponse
from oxidehttp.status import HTTPStatus
from oxidehttp.types import Headers


class CacheProtocol(Protocol):
    async def get(self, key: str) -> CachedResponse | None: ...

    async def set(
        self,
        key: str,
        status_code: HTTPStatus,
        headers: Headers,
        data: bytes | memoryview,
        ttl: float,
    ) -> None: ...

    async def delete(self, key: str) -> None: ...

    async def clear(self) -> None: ...

Your implementation can use any storage system, provided it supports these asynchronous operations.