An Async TTL cache with a lock that forces concurrent calls to wait for the first call to complete and use the cached result rather than recompute

async_ttl_cache.py

import asyncio
import time
import functools


def cache_result(seconds):
    """
    Caches result of an async function, depending on its arguments
    
    No support for maxsize, only TTL for now
    Will lock access to the function being wrapped so that if two calls are being made close to 
    eachother (within $seconds), the second invocation will simply wait for the first call's result
    to be ready rather than async start computing the same value. This is unlike the behavior
    of async_cache and onecache
    """
    def decorator(func):
        cache = {}
        lock = asyncio.Lock()
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            async with lock:
                key = (func.__name__, args, frozenset(kwargs.items()))
                if key in cache and time.time() - cache[key]['time'] < seconds:
                    return cache[key]['value']
                else:
                    cache[key] = {'time': time.time(), 'value': await func(*args, **kwargs)}
                    return cache[key]['value']
        return wrapper
    return decorator

Demo usage:

@cache_result(10)
async def expensive():
    print("starting expensive calc")
    value = 10 + 12
    await asyncio.sleep(3)

    return value

async def main():

    start = time.perf_counter()

    task1 = asyncio.create_task(expensive())
    task2 = asyncio.create_task(expensive())

    print(await task1)

    task3 = asyncio.create_task(expensive())

    print(await task2)
    print(await task3)

    print(f"took {time.perf_counter() - start:.2f}s")


if __name__ == '__main__':
    asyncio.run(main())

Will print

starting expensive calc
22
22
22
took 3.00s

Rather than ~6 seconds with onecache or async-cache.