"""Layered cache combining fast memory and durable disk storage."""
from __future__ import annotations
import asyncio
from typing import Any, Mapping
from pydantic import Field, ValidationError
from accuralai_core.contracts.errors import CacheError
from accuralai_core.contracts.models import GenerateRequest, GenerateResponse
from .base import BaseCache, CacheConfig, extract_options
from .disk import build_disk_cache
from .memory import build_memory_cache
[docs]
class LayeredCacheOptions(CacheConfig):
"""Configuration payload for the layered cache."""
memory: dict[str, Any] = Field(default_factory=dict)
disk: dict[str, Any] = Field(default_factory=dict)
promote_on_hit: bool = True
[docs]
class LayeredCache(BaseCache):
"""Compose two cache implementations into an L1/L2 hierarchy."""
[docs]
def __init__(
self,
*,
memory_cache,
disk_cache,
options: LayeredCacheOptions | None = None,
) -> None:
super().__init__(options=options)
self.options: LayeredCacheOptions = options or LayeredCacheOptions()
self._memory = memory_cache
self._disk = disk_cache
[docs]
async def get(self, key: str, *, request: GenerateRequest) -> GenerateResponse | None:
# First try the in-memory layer.
result = await self._memory.get(key, request=request)
if result is not None:
self._mark_hit()
return self._clone(result)
# Fallback to disk. Promote on hit to keep hot items fast.
disk_result = await self._disk.get(key, request=request)
if disk_result is None:
self._mark_miss()
return None
if self.options.promote_on_hit:
await self._memory.set(key, disk_result)
self._mark_hit()
return self._clone(disk_result)
[docs]
async def set(self, key: str, value: GenerateResponse, *, ttl_s: int | None = None) -> None:
await asyncio.gather(
self._memory.set(key, value, ttl_s=ttl_s),
self._disk.set(key, value, ttl_s=ttl_s),
)
[docs]
async def invalidate(self, key: str) -> None:
await asyncio.gather(self._memory.invalidate(key), self._disk.invalidate(key))
[docs]
async def invalidate_prefix(self, prefix: str) -> None:
await asyncio.gather(
self._memory.invalidate_prefix(prefix),
self._disk.invalidate_prefix(prefix),
)
[docs]
async def build_layered_cache(
*,
config: Mapping[str, Any] | Any | None = None,
**_: Any,
) -> LayeredCache:
"""Factory constructing the layered cache."""
try:
options, ttl_override = extract_options(config=config, options_model=LayeredCacheOptions)
except ValidationError as error:
raise CacheError(f"Invalid layered cache options: {error}") from error
memory_options = dict(options.memory)
disk_options = dict(options.disk)
if ttl_override is not None:
memory_options.setdefault("default_ttl_s", float(ttl_override))
disk_options.setdefault("default_ttl_s", float(ttl_override))
memory_cache = await build_memory_cache(config=memory_options)
disk_cache = await build_disk_cache(config=disk_options)
return LayeredCache(memory_cache=memory_cache, disk_cache=disk_cache, options=options)