Pydantic Cache
Cache results of Python functions, with support for serialization of rich data types powered by Pydantic.
Supports caching to disk or Redis by default, but additional caching backends can easily be added.
Installation
pip install pydantic-cacheUsage
Basic usage
You can use any data types which can be serialized by Pydantic, both in the function signature (cache key) and the returned values:
from datetime import datetime, timedelta
from pydantic import BaseModel
from pydantic_cache import disk_cache
class MyModel(BaseModel):
a: int
b: datetime
d: set[datetime]
@disk_cache(path="~/.cache/my-function", ttl=timedelta(days=1))
def my_function(date: datetime) -> list[MyModel]:
return [] # Some expensive computationIn the above example, subsequent calls to the function with the same argument will fetch the results from the cache on disk. Serialization and deserialization are handled based on the function's type annotations.
Redis support
The library includes support for caching results to/from redis. This depends on redis, which can be installed via pip install pydantic-cache[redis].
from datetime import timedelta
from pydantic_cache import cache
from pydantic_cache.backend import RedisBackend
from redis import Redis
redis = Redis(...)
@cache(RedisBackend(redis, ttl=timedelta(days=1)))
def my_function() -> dict:
return {}Custom cache backends
You can implement custom cache backends by sub-classing Backend:
from pydantic_cache import Backend, cache
class MemoryBackend(Backend):
def __init__(self) -> None:
# Optional initial set-up of the backend.
self._cache: dict[str, str] = {}
def get(self, key: str) -> str:
# Implement cache retrieval here.
# Cache misses should raise a KeyError.
return self._cache[key]
def write(self, key: str, value: str) -> None:
# Write to the cache here.
self._cache[key] = value
@cache(backend=MemoryBackend)
def my_function() -> dict:
return {}Note
Cache backends only interact with serialized data, so the str types above will apply for all backends.
Deferred backend resolution
Some backends may rely on reading configurations or creating connections to external services, which is best avoided at import time. To support this, the cache decorator optionally accepts a callable which returns the backend, instead of the backend itself.
from datetime import timedelta
from pathlib import Path
from pydantic_cache import DiskBackend, cache
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
cache_ttl: timedelta
cache_path: Path
def get_cache_backend() -> DiskBackend:
settings = Settings()
return DiskBackend(settings.cache_path, ttl=settings.cache_ttl)
@cache(backend=get_cache_backend)
def my_function() -> dict:
return {}
asyncio support
Asynchronous functions are supported by default, however using a synchronous backend will naturally result in blocking calls:
import asyncio
from pydantic_cache import DiskBackend, cache
@cache(backend=DiskBackend(...))
async def my_function() -> dict:
return await asyncio.sleep(0, {})To avoid blocking IO calls in the cache backend, you can implement an asynchronous backend as a subclass of AsyncBackend. See the following example using aioredis:
import asyncio
import aioredis
from pydantic_cache import AsyncBackend, cache
class AioRedisBackend(AsyncBackend):
def __init__(self, redis: aioredis.Client) -> None:
self.redis = redis
async def get(self, key: str) -> str:
result = await self.redis.get(key)
if result is None:
raise KeyError(key)
return result
async def write(self, key: str, value: str) -> None:
await self.redis.set(key, value)
@cache(backend=AioRedisBackend(...))
async def my_function() -> dict:
return await asyncio.sleep(0, {})Development
Install dependencies:
pyenv shell 3.10.x
pre-commit install # Configure commit hooks
poetry install # Install Python dependenciesRun tests:
poetry run inv verifyLicense
This project is distributed under the MIT license.
