Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

Python clearing lru_cache every defined amount of time

I am making many redundant calls to an api, so I decided to utilize lru_cache. However, I also wanted to clear the cache every so often in case the response value changes for a given input. So as a solution I created my own decorator in order to clear the cache every given amount of time, but now my function api_call() always seems to return None no matter what. Here's my code:

def cache_maintainer(clear_time: int):
    """
    A function decorator that clears lru_cache clear_time seconds
    :param clear_time: In seconds, how often to clear cache (only checks when called)
    """
    def inner(func):
        def wrapper(*args, **kwargs):
            if hasattr(func, 'next_clear'):
                if time.time() > func.next_clear:
                    func.cache_clear()
                    func.next_clear = time.time() + clear_time
            else:
                func.next_clear = time.time() + clear_time

            func(*args, **kwargs)
        return wrapper
    return inner


@cache_maintainer(86400)  # clear cache every 1 day
@lru_cache(maxsize=1000)
def api_call(param):
    return param

print(api_call("test"))

I can't for the life of me figure out what I've done wrong

like image 957
Themis Avatar asked Feb 16 '26 23:02

Themis


2 Answers

Very simple mistake, took much too long to figure out. As Tim Peters mentioned, func() is called but the result is thrown away (face palm). All I needed to do was add a return where func() is called and it works as I expected. Here's the modified properly working code:

def cache_maintainer(clear_time: int):
    """
    A function decorator that clears lru_cache clear_time seconds
    :param clear_time: In seconds, how often to clear cache (only checks when called)
    """
    def inner(func):
        def wrapper(*args, **kwargs):
            if hasattr(func, 'next_clear'):
                if time.time() > func.next_clear:
                    func.cache_clear()
                    func.next_clear = time.time() + clear_time
            else:
                func.next_clear = time.time() + clear_time

            return func(*args, **kwargs)
        return wrapper
    return inner


@cache_maintainer(86400)  # clear cache every 1 day
@lru_cache(maxsize=1000)
def api_call(param):
    return param

print(api_call("test"))
like image 176
Themis Avatar answered Feb 21 '26 03:02

Themis


This answer is using python package - cachetools

To install cachetools - pip install cachetools

from cachetools import TTLCache

cache = TTLCache(maxsize=5, ttl=100)
cache['dog'] = 'dog'
a = cache['dog'] # dog
... after 100 seconds...
cache['dog']
KeyError exception thrown
like image 21
bigbounty Avatar answered Feb 21 '26 02:02

bigbounty



Donate For Us

If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!