Cache¶

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

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

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

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

class MemoryCache:
    __slots__ = ('_entries', '_lock')

    def __init__(self) -> None:
        self._entries: dict[str, _CacheEntry] = {}
        self._lock = Lock()

    async def get(self, key: str) -> CachedResponse | None:
        async with self._lock:
            entry = self._entries.get(key)

            if entry is None:
                return None

            if monotonic() >= entry.expires_at:
                del self._entries[key]
                return None

            return CachedResponse(
                _status_code=entry.status_code,
                _headers=entry.headers,
                data=bytes(entry.data),
            )

    async def set(
        self,
        key: str,
        status_code: HTTPStatus,
        headers: Headers,
        data: bytes | memoryview,
        ttl: float,
    ) -> None:
        if ttl <= 0:
            return

        entry = _CacheEntry(
            status_code=status_code,
            headers=headers,
            data=data,
            expires_at=monotonic() + ttl,
        )

        async with self._lock:
            self._entries[key] = entry

    async def delete(self, key: str) -> None:
        async with self._lock:
            self._entries.pop(key, None)

    async def clear(self) -> None:
        async with self._lock:
            self._entries.clear()

    async def __len__(self) -> int:
        async with self._lock:
            return len(self._entries)

class RedisCache:
    __slots__ = ('_prefix', '_redis')

    def __init__(
        self, redis: Redis, *, prefix: str = 'oxidehttp:cache:'
    ) -> None:
        self._redis = redis
        self._prefix = prefix.rstrip(':')

    def _key(self, key: str) -> str:
        return f'{self._prefix}:{key}' if self._prefix else key

    async def get(self, key: str) -> CachedResponse | None:
        data = await self._redis.get(self._key(key))

        if data is None:
            return None

        payload = decode(data)

        return CachedResponse(
            _status_code=HTTPStatus(payload['status_code']),
            _headers=payload['headers'],
            data=bytes.fromhex(payload['body']),
        )

    async def set(
        self,
        key: str,
        status_code: HTTPStatus,
        headers: Headers,
        data: bytes | memoryview,
        ttl: float,
    ) -> None:
        if ttl <= 0:
            return

        payload = {
            'status_code': status_code,
            'headers': dict(headers),
            'body': data.hex(),
        }

        data = encode(payload)

        await self._redis.set(self._key(key), data, px=max(1, int(ttl * 1000)))

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

    async def clear(self) -> None:
        pattern = f'{self._prefix}*'

        cursor = 0

        while True:
            cursor, keys = await self._redis.scan(
                cursor=cursor, match=pattern, count=100
            )

            if keys:
                await self._redis.delete(*keys)

            if cursor == 0:
                break