async-lru: Asynchronous LRU Cache for asyncio

Warnings

• breaking Cross-event loop cache access behavior changed significantly between v2.2.0 and v2.3.0. Prior to v2.3.0 (from v2.2.0 onwards), attempting to use an `alru_cache` instance with a different event loop than where it was first called would raise a `RuntimeError` ('alru_cache is not safe to use across event loops').
  Fix: For versions before 2.3.0, ensure a cache instance is strictly used with a single event loop, or create separate cache instances per loop. If you need cross-loop usage, upgrade to v2.3.0+ and be aware of the new auto-reset behavior.
• gotcha As of v2.3.0, cross-event loop cache access no longer raises a `RuntimeError` but instead triggers an auto-reset and rebind to the current event loop, emitting an `AlruCacheLoopResetWarning`. While this prevents hard crashes, it means the cache effectively clears and reinitializes when the event loop changes, potentially losing cached data.
  Fix: If multi-event loop usage is intentional, be mindful that the cache will reset. For persistent caching across different loops or threads, consider explicit cache management (e.g., using `threading.local` for per-thread/loop caches) or a shared, thread-safe external caching mechanism.
• gotcha It is highly recommended to explicitly close `alru_cache` instances using `cache_close()`, especially when using `ttl` (time-to-live). Failing to close the cache can lead to resource leaks (e.g., lingering asyncio tasks or timers) that might prevent your application from shutting down cleanly or lead to unexpected behavior.
  Fix: Always call `await func.cache_close()` on your cached functions when they are no longer needed, typically during application shutdown or when disposing of objects that hold cached methods.
• gotcha When using `ttl` (time-to-live) for cache entries, many entries expiring simultaneously can lead to a 'thundering herd' problem, where many clients try to recompute the same value at once. This can negate the benefits of caching and strain backend resources.
  Fix: Utilize the `jitter` parameter (introduced in v2.2.0) with `ttl` to randomize expiration times. For example, `@alru_cache(ttl=3600, jitter=1800)` will spread expirations over a 1.5-hour window around the 1-hour TTL.

Install

• pip install async-lru Install latest version

Imports

• alru_cache

  from async_lru import alru_cache

  Using `functools.lru_cache` on an `async def` function will cache the coroutine object itself, not its awaited result, leading to `RuntimeError: cannot reuse already awaited coroutine` on subsequent awaits. `async-lru` provides `alru_cache` for correct async function caching.

Quickstart

This quickstart demonstrates basic usage of the `alru_cache` decorator with `maxsize`, `ttl` (time-to-live), and `jitter` parameters. It shows how to inspect cache statistics using `cache_info()`, check for cache presence with `cache_contains()`, and explicitly close the cache with `cache_close()` to release resources. Note that `aiohttp` is used for demonstration purposes of an actual async network call.

import asyncio
import aiohttp
from async_lru import alru_cache

@alru_cache(maxsize=32, ttl=10, jitter=2)
async def get_pep(num):
    """Fetches a PEP from python.org, caches the result."""
    resource = f'http://www.python.org/dev/peps/pep-{num:04d}/'
    print(f"Fetching PEP {num}...")
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(resource) as s:
                if s.status == 200:
                    return await s.text()
                return f'Not Found (Status: {s.status})'
        except aiohttp.ClientError as e:
            return f'Network Error: {e}'

async def main():
    print("\n--- First round (misses) ---")
    for n in 8, 290, 308, 320:
        pep = await get_pep(n)
        print(f"PEP {n}: {len(pep) if pep else 'Error'} characters")

    print("\n--- Second round (hits) ---")
    for n in 8, 218, 320:
        pep = await get_pep(n)
        print(f"PEP {n}: {len(pep) if pep else 'Error'} characters")

    print("\n--- Cache Info ---")
    print(get_pep.cache_info())

    print("\n--- Checking cache_contains ---")
    print(f"Cache contains PEP 8: {get_pep.cache_contains(8)}")
    print(f"Cache contains PEP 9991: {get_pep.cache_contains(9991)}")

    # Simulate passage of time for TTL
    print("\n--- Waiting for TTL expiration (10 seconds) ---")
    await asyncio.sleep(10) # Wait for TTL

    print("\n--- After TTL: PEP 8 (should re-fetch) ---")
    pep = await get_pep(8)
    print(f"PEP 8: {len(pep) if pep else 'Error'} characters")
    print(get_pep.cache_info())

    # Closing is optional but highly recommended to release resources
    await get_pep.cache_close()

if __name__ == '__main__':
    # This example requires aiohttp for network requests
    # If aiohttp is not installed, the example will still run but 'get_pep' will fail.
    # pip install aiohttp
    try:
        asyncio.run(main())
    except RuntimeError as e:
        print(f"Caught a runtime error: {e}. This might happen if the event loop is already running.")