User Guide

Installation

pip install async-cache

Requires Python 3.8+. No external dependencies.

When to Use Which Feature

Scenario	Feature	Why
Simple function memoization	`@AsyncLRU`	Decorator, LRU eviction, no TTL needed
Time-sensitive data (sessions, tokens)	`@AsyncTTL`	Automatic expiry after `time_to_live` seconds
Fine-grained key/TTL control	`AsyncCache` (function API)	Per-key TTL, manual set/get/delete, batch loader
High-concurrency hot keys	`AsyncCache` with `loader`	Built-in thundering herd protection
GraphQL resolvers / N+1	`AsyncCache` with `batch_loader`	DataLoader-style batching within a time window
Survive process restarts	`DiskBackend`	Pickle-based persistence, zero new dependencies
Distributed / multi-instance cache	`RedisBackend`	Two-tier L1/L2 caching with built-in pure-Python Redis client
AI agent tool caching	`AgentCache` / `AgentCacheSession`	Resource tagging, session scoping, loop detection
Write-through invalidation	`AsyncLRUInvalidator` / `AsyncTTLInvalidator` / `AgentCacheInvalidator`	Decorator-based, ties mutations to cached reads; invalidates before mutation

Quick Start

Function API

import asyncio
from cache import AsyncCache

cache = AsyncCache(maxsize=1000, default_ttl=300)

async def get_user(user_id: int) -> dict:
    return await cache.get(
        f"user:{user_id}",
        loader=lambda: fetch_from_database(user_id),
    )

# Warmup hot keys at startup
await cache.warmup({"hot:key": lambda: preload_hot()})

# Observability
print(cache.get_metrics())  # {'hits': 950, 'misses': 50, 'size': 200, 'hit_rate': 0.95}

Decorator API

from cache import AsyncLRU, AsyncTTL

@AsyncLRU(maxsize=128)
async def get_product(product_id: int):
    return await db.query_product(product_id)

@AsyncTTL(time_to_live=60, skip_args=1)  # skip 'self'
async def get_session(self, session_id: str):
    return await db.get_session(session_id)

Core Features

Thundering Herd Protection

When a cached item expires under heavy load, multiple concurrent requests would normally trigger duplicate backend calls. async-cache ensures only one loader executes while all others await the same result.

cache = AsyncCache()

async def loader():
    return await expensive_db_query()  # runs ONCE

# 100 concurrent coroutines, 1 database call
results = await asyncio.gather(*[
    cache.get("hot_key", loader=loader) for _ in range(100)
])

DataLoader-Style Batching

Group concurrent gets into a single batch call to reduce database round-trips. Configurable window (default 5ms) and batch size (default 100).

async def batch_loader(keys):
    return await db.batch_query(keys)  # one query for all keys

# Auto-batched within 5ms window
user1, user2 = await asyncio.gather(
    cache.get(1, batch_loader=batch_loader),
    cache.get(2, batch_loader=batch_loader),
)

Cache Warmup

Preload critical data at startup to avoid cold-start latency spikes.

await cache.warmup({
    "config:app": load_app_config,
    "feature_flags": load_feature_flags,
    "popular:products": load_popular_products,
})

Metrics & Observability

metrics = cache.get_metrics()
# {'hits': 950, 'misses': 50, 'size': 200, 'hit_rate': 0.95}

# Decorator metrics
metrics = get_product.get_metrics()

TTL & Manual Invalidation

cache = AsyncCache(default_ttl=300)         # global 5-min default
cache.set("session:123", data, ttl=3600)    # per-key 1 hour
cache.set("permanent", data, ttl=None)      # no expiration

cache.delete("session:123")                 # single key
cache.clear()                               # all keys

Force Refresh

Bypass cache for a single call using use_cache=False:

@AsyncTTL(time_to_live=60)
async def get_status():
    return await check_service_status()

status = await get_status(use_cache=False)  # forces fresh load

Skip Arguments in Cache Key

For instance methods, skip self from the cache key:

class UserService:
    @AsyncLRU(maxsize=100, skip_args=1)
    async def get_user(self, user_id: int):
        return await self.db.get_user(user_id)

Configuration Reference

AsyncCache Parameters

Parameter	Default	Description
`maxsize`	128	Maximum items in cache (`None` = unlimited)
`default_ttl`	None	Default TTL in seconds (`None` = no expiration)
`batch_window_ms`	5	Batch grouping window in milliseconds
`max_batch_size`	100	Max keys per batch call
`backup`	None	Optional `CacheBackend` for disk persistence (e.g., `DiskBackend`)
`remote_cache`	None	Optional `RemoteCacheBackend` for distributed L2 caching (e.g., `RedisBackend`)

AsyncLRU / AsyncTTL Parameters

Parameter	Default	Description
`maxsize`	128 / 1024	Maximum cache size
`time_to_live` (TTL only)	60	TTL in seconds
`skip_args`	0	Number of leading args to skip in cache key
`backup`	None	Optional `CacheBackend` for disk persistence
`remote_cache`	None	Optional `RemoteCacheBackend` for distributed L2 caching