Python lru_cache with timeout

timed_cache.py

from datetime import datetime, timedelta
import functools


def timed_cache(**timedelta_kwargs):                                              
                                                                                  
    def _wrapper(f):                                                              
        update_delta = timedelta(**timedelta_kwargs)                              
        next_update = datetime.utcnow() + update_delta                            
        # Apply @lru_cache to f with no cache size limit                          
        f = functools.lru_cache(None)(f)                                          
                                                                                                                      
        @functools.wraps(f)                                                       
        def _wrapped(*args, **kwargs):                                            
            nonlocal next_update                                                  
            now = datetime.utcnow()                                               
            if now >= next_update:                                                
                f.cache_clear()                                                   
                next_update = now + update_delta                                
            return f(*args, **kwargs)                                             
        return _wrapped                                                           
    return _wrapper

many thanks to everybody sharing here! to further pile on to this gist, here are my suggested changes to @svpino's version:

def lru_cache(timeout: int, maxsize: int = 128, typed: bool = False):
    def wrapper_cache(func):
        func = functools.lru_cache(maxsize=maxsize, typed=typed)(func)
        func.delta = timeout * 10 ** 9
        func.expiration = time.monotonic_ns() + func.delta

        @functools.wraps(func)
        def wrapped_func(*args, **kwargs):
            if time.monotonic_ns() >= func.expiration:
                func.cache_clear()
                func.expiration = time.monotonic_ns() + func.delta
            return func(*args, **kwargs)

        wrapped_func.cache_info = func.cache_info
        wrapped_func.cache_clear = func.cache_clear
        return wrapped_func
    return wrapper_cache

• renamed the decorator to lru_cache and the timeout parameter to timeout ;)
• using time.monotonic_ns avoids expensive conversion to and from datetime/timedelta and prevents possible issues with system clocks drifting or changing
• attaching the original lru_cache's cache_info and cache_clear methods to our wrapped_func

Solid update, @fdemmer.

Further tidying up from @fdemmer version, a fully working snippet

from functools import lru_cache, wraps
from time import monotonic_ns


def timed_lru_cache(
    _func=None, *, seconds: int = 600, maxsize: int = 128, typed: bool = False
):
    """Extension of functools lru_cache with a timeout

    Parameters:
    seconds (int): Timeout in seconds to clear the WHOLE cache, default = 10 minutes
    maxsize (int): Maximum Size of the Cache
    typed (bool): Same value of different type will be a different entry

    """

    def wrapper_cache(f):
        f = lru_cache(maxsize=maxsize, typed=typed)(f)
        f.delta = seconds * 10 ** 9
        f.expiration = monotonic_ns() + f.delta

        @wraps(f)
        def wrapped_f(*args, **kwargs):
            if monotonic_ns() >= f.expiration:
                f.cache_clear()
                f.expiration = monotonic_ns() + f.delta
            return f(*args, **kwargs)

        wrapped_f.cache_info = f.cache_info
        wrapped_f.cache_clear = f.cache_clear
        return wrapped_f

    # To allow decorator to be used without arguments
    if _func is None:
        return wrapper_cache
    else:
        return wrapper_cache(_func)

With documentations, imports, and allow decorators to be called without arguments and paratheses

@jianshen92 👌💪

The implementation has a big problem: if you have a function that you can call with different values and you obviously want the result cached with TTL for each calling value, then when the TTL is reached for one calling value, the cache is cleared of ALL CACHED RESULTS, that is FOR ALL CALLING VALUES.

Sample code:

import time
import random

@timed_cache(seconds=10)
def expensive_operation(a: int):
    return random.randint(1, 1 + a)
    
def ex_op_wrapper(a: int):
    return f'{time.time()}: {expensive_operation(a)}'

Calling in reply with a 6 secs pause between the first and second call:

ex_op_wrapper(1000)
'1657014039.3334417: 762'
ex_op_wrapper(100)
'1657014045.5532942: 4'
ex_op_wrapper(1000)
'1657014047.3158472: 762'
ex_op_wrapper(100)
'1657014048.6246898: 4'
ex_op_wrapper(1000)
'1657014049.7079725: 847'
ex_op_wrapper(100)
'1657014050.7649162: 70'

You can see that the first cached result for calling with 100 was 4 at '1657014045.5532942', then that was changed at '1657014050.7649162' to 70, so only 5 secs after the first caching of 4, instead of 10.

The problem in the above code is that f.cache_clear() clears the cache for all calling values, not just for the expired one.

Thanks guys ! Btw it can leads to a TypeError: unhashable type: 'list' if you have list args.
A fix could be to cast those args to tuple (more info : https://stackoverflow.com/a/49210802)
This piece of code will fix this :

for arg, value in kwargs.items():
    kwargs[arg] = tuple(value) if type(value) == list else value

Thanks guys ! Btw it can leads to a TypeError: unhashable type: 'list' if you have list args. A fix could be to cast those args to tuple (more info : https://stackoverflow.com/a/49210802) This piece of code will fix this :

for arg, value in kwargs.items():
    kwargs[arg] = tuple(value) if type(value) == list else value

The behavior remains the same but I would suggest to use isinstance() instead of type()

for arg, value in kwargs.items():
    kwargs[arg] = tuple(value) if isinstance(value, list) else value

Thanks for the implementations! Really helpful!

Something I noticed is that neither of these implementations work with pytest-antilru. This is likely due to the lru_cache which is monkeypatched is not patched early enough: ipwnponies/pytest-antilru#28.

args = [tuple(v) if isinstance(v, list) else v for v in args]

for args too

Thanks guys ! Btw it can leads to a TypeError: unhashable type: 'list' if you have list args. A fix could be to cast those args to tuple (more info : https://stackoverflow.com/a/49210802) This piece of code will fix this :

for arg, value in kwargs.items():
    kwargs[arg] = tuple(value) if type(value) == list else value

The behavior remains the same but I would suggest to use isinstance() instead of type()

for arg, value in kwargs.items():
    kwargs[arg] = tuple(value) if isinstance(value, list) else value