Async LRU/LFU caching

Cache async function results with LRU or LFU eviction, optional TTL, and built-in profiling statistics.

The cache module provides AsyncCache — a thread-safe, async-native cache — and the @async_cache decorator that wraps any coroutine function with automatic caching.

Installation

pip install alex-ber-utils

async_cache decorator

The simplest way to add caching is with the @async_cache decorator.

from alexber.utils.cache import async_cache

@async_cache(maxsize=256, policy="LRU", ttl=60)
async def fetch_user(user_id: int) -> dict:
    # Expensive I/O call — only runs on a cache miss
    return await db.get_user(user_id)

Calling fetch_user(42) twice within 60 seconds returns the cached result on the second call.

Accessing the cache instance

The decorator attaches the underlying AsyncCache instance to the wrapped function:

stats = await fetch_user.cache_instance.get_stats()
print(stats)
# {
#   'hits': 5, 'misses': 1, 'hit_miss_ratio': '0.8333',
#   'avg_time': '0.0023 s', 'max_time': '0.0041 s',
#   'min_time': '0.0019 s', 'total_calls': 1,
#   'current_size': 1, 'max_size': 256, 'ttl_sec': '60.0000'
# }

Caching class methods

The decorator detects bound methods via inspect and keys the cache per instance using id(self), so two instances never share cached results:

from alexber.utils.cache import async_cache

class WeatherService:
    def __init__(self, city: str):
        self.city = city

    @async_cache(maxsize=128, policy="LFU")
    async def get_forecast(self, date: str) -> dict:
        return await weather_api.fetch(self.city, date)

service_a = WeatherService("London")
service_b = WeatherService("Paris")

# These are cached independently — different instance IDs
await service_a.get_forecast("2026-03-25")
await service_b.get_forecast("2026-03-25")

maxsize · int

Maximum number of entries to hold in the cache. When the cache is full, the least-recently-used (LRU) or least-frequently-used (LFU) entry is evicted. Required when using AsyncCache directly; optional in @async_cache.

policy · string · default: "LFU"

Eviction policy. Accepts "LRU" (Least Recently Used) or "LFU" (Least Frequently Used). LFU keeps the entries called most often; LRU keeps the entries called most recently.

ttl · float | None · default: "None"

Time-to-Live in seconds. After this many seconds an entry is treated as expired and the next access triggers a fresh call. None disables expiry.

---

AsyncCache class

Use AsyncCache directly when you need fine-grained control — for example, to share one cache across multiple functions or to call clear() programmatically.

import asyncio
from alexber.utils.cache import AsyncCache

cache = AsyncCache(maxsize=512, policy="LRU", ttl=300)

async def get_price(ticker: str) -> float:
    try:
        return await cache[ticker]
    except KeyError:
        start = ...
        price = await market_api.quote(ticker)
        exec_ns = ...  # measure elapsed nanoseconds
        await cache.update_profiling(exec_ns)
        await cache.__setitem__(ticker, price)
        return price

Constructor parameters

maxsize · int · required

Maximum number of entries the cache will hold before evicting.

policy · string · default: "LFU"

Eviction policy: "LRU" or "LFU".

ttl · float | None · default: "None"

Entry lifetime in seconds. None means entries never expire.

Methods

Method	Description
await cache[key]	Return cached value. Raises KeyError on miss or expiry.
await cache.__setitem__(key, value)	Store a value. Sets expiry if TTL is configured.
await cache.update_profiling(exec_time_ns)	Record an execution time sample (nanoseconds).
await cache.get_stats()	Return a dict of hit/miss/timing statistics (see below).
await cache.clear()	Evict all entries and reset all statistics.

Statistics returned by get_stats()

Key	Type	Description
hits	int	Number of successful cache lookups.
misses	int	Number of cache misses.
hit_miss_ratio	str	hits / (hits + misses), formatted to 4 decimal places.
avg_time	str	Average execution time across all recorded calls.
max_time	str	Longest recorded execution time.
min_time	str	Shortest recorded execution time.
total_calls	int	Number of calls passed to update_profiling.
current_size	int	Number of entries currently in the cache.
max_size	int	Configured maxsize.
ttl_sec	str	Configured TTL (or "None").

---

Eviction policies

LRU — Least Recently Used

Evicts the entry that was accessed least recently. Good when access patterns are temporal — recently requested data is likely to be requested again soon.

LFU — Least Frequently Used

Evicts the entry with the lowest access count. Good when some data is accessed far more often than the rest and should be protected from eviction.

---

Cache key internals

The cache automatically converts unhashable argument values (lists, dicts, custom objects) into stable cache keys using two helpers from mains.py.

make_hashable(obj)

Recursively converts an object to a hashable type:

• Mappings → frozenset of (key, value) pairs
• Iterables (excluding str / bytes) → tuple
• Natively hashable objects → returned as-is
• Everything else → wrapped in HashableWrapper

from alexber.utils.mains import make_hashable

make_hashable({"a": [1, 2], "b": 3})
# frozenset({('a', (1, 2)), ('b', 3)})

make_hashable([1, {"x": 2}])
# (1, frozenset({('x', 2)}))

HashableWrapper

A fallback wrapper for objects that are inherently non-hashable. It derives its hash from str(obj) and compares by identity of the wrapped value.

is_iterable(value) and is_mapping(value)

Two guard utilities used by make_hashable internally:

from alexber.utils.mains import is_iterable, is_mapping

is_iterable([1, 2, 3])   # True
is_iterable("hello")     # False  — str is excluded
is_iterable(None)        # False

is_mapping({"a": 1})     # True
is_mapping(["a", 1])     # False

is_iterable explicitly excludes str and bytes so that string arguments are treated as atomic cache key components rather than sequences of characters.

---

Full end-to-end example

1. Install the package

   pip install alex-ber-utils
   
2. Decorate your async function

   import asyncio
   from alexber.utils.cache import async_cache
   
   @async_cache(maxsize=100, policy="LFU", ttl=30)
   async def compute(x: int, y: int) -> int:
       await asyncio.sleep(0.1)  # simulate work
       return x + y
   
3. Call the function and inspect statistics

   async def main():
       print(await compute(1, 2))  # miss — runs the function
       print(await compute(1, 2))  # hit  — returns cached result
       print(await compute(3, 4))  # miss
   
       stats = await compute.cache_instance.get_stats()
       print(stats['hits'])         # 1
       print(stats['misses'])       # 2
       print(stats['hit_miss_ratio'])  # '0.3333'
   
   asyncio.run(main())
   
4. Clear the cache when needed

   await compute.cache_instance.clear()
   

Use policy="LFU" (the default) when a small subset of arguments dominates your call traffic. Switch to policy="LRU" when you expect a sliding window of recent queries with little repetition.