Cache Invalidation

async-cache provides invalidation at every API level: core AsyncCache, decorators (AsyncLRU / AsyncTTL), and agent cache (AgentCacheInvalidator).

Core AsyncCache

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

# Delete a single key
cache.delete("user:123")

# Clear all entries and reset metrics
cache.clear()

Decorator API: Per-Key Invalidation

Every @AsyncLRU and @AsyncTTL wrapper exposes invalidate_cache():

from cache import AsyncLRU

@AsyncLRU(maxsize=128)
async def get_user(user_id: int):
    return await db.get_user(user_id)

# Invalidate a specific cached result
get_user.invalidate_cache(42)  # removes cache entry for get_user(42)

# Clear the entire cache
get_user.clear_cache()

Invalidation Decorators

For write/mutation functions that should automatically invalidate related cached reads, use AsyncLRUInvalidator or AsyncTTLInvalidator.

Important: Invalidation happens before the mutation executes. This ensures no reader ever sees stale data after a mutation starts, even if the mutation raises an exception.

AsyncLRUInvalidator

from cache import AsyncLRU, AsyncLRUInvalidator

@AsyncLRU(maxsize=128)
async def get_user(user_id: int):
    return await db.get_user(user_id)

@AsyncLRUInvalidator(get_user)
async def update_user(user_id: int, data: dict):
    await db.update_user(user_id, data)

await get_user(1)              # miss -> loads from DB
await get_user(1)              # hit
await update_user(1, {"name": "Alice"})  # invalidates get_user(1), then runs mutation
await get_user(1)              # miss -> reloads fresh data

AsyncTTLInvalidator

from cache import AsyncTTL, AsyncTTLInvalidator

@AsyncTTL(time_to_live=60)
async def get_session(session_id: str):
    return await db.get_session(session_id)

@AsyncTTLInvalidator(get_session)
async def end_session(session_id: str):
    await db.delete_session(session_id)

await get_session("s1")    # miss -> loads
await end_session("s1")    # invalidates get_session("s1"), then runs mutation
await get_session("s1")    # miss -> reloads

Invalidator Options

Parameter	Default	Description
`cached_fn`	(required)	The `@AsyncLRU` / `@AsyncTTL` wrapped function to invalidate
`clear_all`	False	Clear the entire cache instead of a specific key
`key_fn`	None	Custom `(args, kwargs) -> (inv_args, inv_kwargs)` for complex arg mapping

Note

skip_args has been removed from invalidators. Passing a non-zero value raises TypeError. The old skip_args on invalidators silently produced wrong cache keys when the cached function also used skip_args. Use key_fn instead — see the example below.

Mutation Failure Safety

The cache key is evicted before the mutation runs. If the mutation raises, the key is already gone — the next read triggers a fresh load:

@AsyncLRUInvalidator(get_user)
async def update_user(user_id: int, data: dict):
    raise RuntimeError("DB error")  # mutation fails

await get_user(1)                       # loads and caches
try:
    await update_user(1, {"x": "y"})   # invalidates first, then raises
except RuntimeError:
    pass
await get_user(1)                       # cache miss -> reloads fresh data

Clear All Mode

When a mutation affects all cached entries:

@AsyncLRUInvalidator(get_user, clear_all=True)
async def rebuild_user_index():
    await db.rebuild_index()

await rebuild_user_index()  # clears ALL get_user cache entries

Custom Key Mapping (`key_fn`)

When the mutation function’s args don’t directly match the cached function’s args, use key_fn to translate them:

@AsyncLRU(maxsize=128)
async def get_item(item_id: int):
    return await db.get_item(item_id)

# Mutation has extra args; extract just item_id for invalidation
@AsyncLRUInvalidator(get_item, key_fn=lambda args, kw: (args[:1], {}))
async def update_item(item_id: int, name: str, price: float):
    await db.update_item(item_id, name=name, price=price)

Migrating from ``skip_args``: if you previously used skip_args=1 on the invalidator to skip self, replace it with key_fn:

# Old (raises TypeError now):
# @AsyncLRUInvalidator(get_data, skip_args=1)

# New:
@AsyncLRUInvalidator(get_data, key_fn=lambda args, kw: (args[1:], {}))
async def update_data(self, key):
    ...

AgentCache Invalidation

For AI agent workflows, use resource-based invalidation. Like the decorator invalidators, AgentCacheInvalidator evicts the cache before the mutation runs — failed mutations never leave stale data.

from agent_cache import AgentCache, AgentCacheInvalidator

@AgentCache(resource="cart", scope="global", ttl=60)
async def get_cart(user_id):
    return await db.get_cart(user_id)

@AgentCacheInvalidator(resource="cart", scope="global")
async def add_to_cart(user_id, item):
    await db.add_item(user_id, item)

await get_cart("u1")              # cached
await add_to_cart("u1", "laptop") # invalidates ALL "cart" entries, then runs mutation
await get_cart("u1")              # re-fetched

AgentCacheInvalidator accepts the same invalidation parameters as the decorator invalidators:

Parameter	Default	Description
`resource`	(required)	Resource tag to invalidate
`scope`	`"global"`	`"global"` or `"session"`
`clear_all`	True	Clear all entries for the resource (default for agent cache)
`key_fn`	None	Custom `(args, kwargs) -> cache_key` for selective invalidation