functool lru_cache with timeout allowing to set expiration time on cached values

lru_cache_timeout.py

import datetime
from functools import lru_cache, wraps



def lru_cache_timeout(timeout_seconds: int):
    def decorator(f):

        @lru_cache(maxsize=None)
        def f_cache_id(f_cache_id, *args, **kwds):
            return f(*args, **kwds)

        @wraps(f)
        def decorated(*args, **kwds):
            cache_id = int(datetime.datetime.now().timestamp() / timeout_seconds)
            return f_cache_id(cache_id, *args, **kwds)

        return decorated

    return decorator


if __name__ == '__main__':

    TIMEOUT_SECONDS = 2

    # get_database_connection() is an example usage
    conn_id = 0
    @lru_cache_timeout(timeout_seconds=TIMEOUT_SECONDS)
    def get_database_connection():
        global conn_id
        conn_id += 1
        print('Created new database connection ID: {}'.format(conn_id))
        return conn_id

    # Retrieve 3 times on the same time, should be the same connection id
    assert get_database_connection() == 1
    assert get_database_connection() == 1
    assert get_database_connection() == 1

    # Sleep to make sure the timeout is gone:

    import time
    time.sleep(TIMEOUT_SECONDS)

    # Connection changed, but still cached
    assert get_database_connection() == 2
    assert get_database_connection() == 2
    assert get_database_connection() == 2