Implementation of an lru_cache() decorator that ignores the first argument

Robert Latest boblatest at yahoo.com
Wed Sep 28 14:22:15 EDT 2022


Hi all,

in a (Flask) web application I often find that many equal (SQLAlchemy) queries
are executed across subsequent requests. So I tried to cache the results of
those queries on the module level like this:

    @lru_cache()
    def query_db(db, args):
        # do the "expensive" query
        return result

This obviously doesn't work because each request uses a new database session,
so the db argument always changes from one call to the next, triggering a new
query against the database. But even if that weren't so, the function would
keep returning the same value forever (unless it's kicked out of the cache) and
not reflect the (infrequent) changes on the database. So what I need is some
decorator that can be used like this:

    @lru_ignore_first(timeout=10)
    def query_db(db, args):
        # do the "expensive" query
        return result

This is what I came up with. I'm quite happy with it so far.  Question: Am I
being too clever? is it too complicated? Am I overlooking something that will
come back and bite me later? Thanks for any comments!

    from functools import wraps, lru_cache
    from time import time, sleep

    def lru_ignore_first(timeout=0, **lru_args):

        class TimeCloak():
            '''All instances compare equal until timeout expires'''
            __slots__ = ('x', 't', 'timeout')

            def __init__(self, timeout):
                self.timeout = timeout
                self.t = 0
                self.x = None

            def __hash__(self):
                return self.t

            def __eq__(self, other):
                return self.t == other.t

            def update(self, x):
                self.x = x
                if self.timeout:
                    t = int(time())
                    if t >= self.t + self.timeout:
                        self.t = t

        cloak = TimeCloak(timeout)

        def decorator(func):

            @lru_cache(**lru_args)
            def worker(cloak, *a, **b):
                return func(cloak.x, *a, **b)

            @wraps(func)
            def wrapped(first, *a, **kw):
                cloak.update(first)
                return worker(cloak, *a, **kw)

            return wrapped

        return decorator

    @lru_ignore_first(3)
    def expensive(first, par):
        '''This takes a long time'''
        print('Expensive:', first, par)
        return par * 2

    for i in range(10):
        r = expensive(i, 100)
        sleep(1)
        print(r)


More information about the Python-list mailing list