How to clear expired values

[tonal]

How to clear expired values from InMemoryBackend?

[gabszs]

FastAPI Cache with Redis Backend

InMemoryBackend Clear Method

The InMemoryBackend class provides a clear method to remove cached items:

class InMemoryBackend(Backend):
    async def clear(self, namespace: Optional[str] = None, key: Optional[str] = None) -> int:
        count = 0
        if namespace:
            keys = list(self._store.keys())
            for store_key in keys:
                if store_key.startswith(namespace):
                    del self._store[store_key]
                    count += 1
        elif key:
            if key in self._store:
                del self._store[key]
                count += 1
        return count

Complete FastAPI Example with Redis Cache

Here's a complete working example using FastAPI with Redis-backed caching via FastAPICache:

from contextlib import asynccontextmanager
from typing import Optional
from fastapi import FastAPI, Depends
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from redis import Redis

# Dependency to get cache instance
async def get_cache() -> RedisBackend:
    return FastAPICache.get_backend()

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize Redis backend
    redis_backend = RedisBackend(Redis.from_url(settings.REDIS_URL))
    app.state.cache = redis_backend
    
    # Initialize FastAPICache
    FastAPICache.init(
        redis_backend,
        expire=settings.CACHE_TTL,
        cache_status_header=settings.CACHE_STATUS_HEADER,
    )
    
    yield
    
    # Cleanup on shutdown
    await app.state.cache.close()

# Create FastAPI app with lifespan
app = FastAPI(lifespan=lifespan)

@app.post("/clear-cache")
async def clear_cache(
    namespace: Optional[str] = None,
    key: Optional[str] = None,
    cache: RedisBackend = Depends(get_cache)
):
    """
    Clear cache entries.
    
    Args:
        namespace: Clear all keys with this namespace prefix
        key: Clear specific key
        cache: Redis backend dependency
    
    Returns:
        Number of cleared entries
    """
    count = await cache.clear(namespace=namespace, key=key)
    return {
        "detail": f"Cache cleared",
        "cleared_count": count
    }

Usage Examples

Clear entire cache

curl -X POST "http://localhost:8000/clear-cache"

Clear by namespace

curl -X POST "http://localhost:8000/clear-cache?namespace=user:"

Clear specific key

curl -X POST "http://localhost:8000/clear-cache?key=user:123"

Configuration

Make sure to set up your settings with the required variables:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    REDIS_URL: str = "redis://localhost:6379"
    CACHE_TTL: int = 3600  # 1 hour
    CACHE_STATUS_HEADER: bool = True

settings = Settings()

[tonal]

class ClearedInMemoryBackend(InMemoryBackend):
    async def clear_expired(self) -> None:
        async with self._lock:
            now = self._now
            for_del = tuple(k for k, v in self._store.items() if v.ttl_ts < now)
            for k in for_del:
                del self._store[k]

[ryanisn]

I like the convenience of InMemoryBackend with no dependency to external services, but I'm concerned with the lack of size control, and inmemory cache will eventually eat up server memory.
Is there a way to schedule this clear function to run hourly?