# Import necessary libraries
from fastcore.xtras import flexicache, time_policy, mtime_policy
# Libraries used in testing cache validity and cache invalidation
from random import randint
from pathlib import Path
from time import sleepExploring flexicache
coding
open-source
tech
An exploration of using flexicache for caching in Python.
Note from Jeremy: I’m thrilled that the legendary Daniel Roy Greenfeld took the time to dig into a very recent addition I made to fastcore:
flexicache. It’s a super useful little tool which nowadays I use all the time. I hope you like it as much as Danny and I do!
When coding in Python really like to use decorators to cache results from functions and methods, often to memory and sometimes to ephemeral stores like memcached. In fact, I’ve worked on and created several cache decorators, including one that influenced the implementation of the @cached_property decorator in Python 3.8.
A cache decorator called flexicache is part of the fastcore library. flexicache allows you to cache in memory results from functions and methods in a flexible way. Besides having an implementation of LRU caching, each use of the decorator can be configured to use one or more cache invalidation policies.
Two policies, time_policy and mtime_policy are used to invalidate the cache based on time and file modification time respectively. The time_policy invalidates the cache after a specified number of seconds, while the mtime_policy invalidates the cache if the file has been modified since the last time it was cached.
Let’s try it out!
Basic usage
Here’s a simple function returning a number between 1 to 1000 that we can show being cached. We’ll use this in all our examples.
def random_func(v):
return randint(1, 1000)
# Assert False as the function is not cached
assert random_func(1) != random_func(1)Time policy
This is how we use the time_policy to cache the function.
@flexicache(time_policy(.1))
def random_func():
return randint(1, 1000)
# assert True as the function is cached
assert random_func() == random_func()Let’s use the sleep function to simulate time between calls to random_func.
result = random_func()
# True as the function is cached
assert result == random_func()
# Sleep for .2 seconds to allow cache to expire
sleep(0.2)
# Assert False as the cache has expired and the function is called again
assert result != random_func()File modification time (mtime_policy)
We’ll try with mtime_policy, checking to see if touching a file invalidates the cache. We’ll use this site’s main.py file as the file to touch.
@flexicache(mtime_policy('../../main.py'))
def random_func():
return randint(1, 1000)
# Assert True as the function is cached
assert random_func() == random_func()Now let’s use the Path.touch() method to touch the file. This will update the file’s modification time to the current time, which should invalidate the cache.
# Call the function to cache the result
result = random_func()
assert result == random_func() # True as the function is cached
# Update the file's modification time, which invalidates the cache
Path('../../main.py').touch()
# Assert False as the cache is invalidated
assert result != random_func() Using multiple policies
A unique feature of flexicache is that you can use multiple policies at the same time. This allows you to combine the benefits of different caching strategies. In this example, we’ll use both time_policy and mtime_policy together. This means that the cache will be invalidated if either the time limit is reached or the file has been modified.
Testing the cache with both policies is identical to the previous examples. We’ll call the function, first with the time policy, then with the mtime policy, and finally with both policies. We’ll also touch the file to see if it invalidates the cache.
@flexicache(time_policy(.1), mtime_policy('../../main.py'))
def random_func():
return randint(1, 1000)
# True as the function is cached
assert random_func() == random_func()Testing time invalidation is the same as before. We’ll call the function, wait for the time limit to be reached, and then call it again to see if the cache is invalidated.
result = random_func()
# True as the function is cached
assert result == random_func()
# Sleep for .2 seconds to allow cache to expire
sleep(0.2)
# False as the cache has expired and the function is called again
assert result != random_func() Testing file timestamp is the same as before. We’ll call the function, touch the file, and then call it again to see if the cache is invalidated.
# Call the function to cache the result
result = random_func()
# True as the function is cached
assert result == random_func()
# Update the file's modification time, which invalidates the cache
Path('../../main.py').touch()
# Assert False as the cache is invalidated
assert result != random_func() What about LRU caching?
Now let’s test out the flexicache decorator to see how it behaves as an lru_cache replacement. For reference, LRU caching is a caching strategy that keeps track of the most recently used items and removes the least recently used items when the cache reaches its maximum size. In other words, it takes out the latest items from the cache first when it runs out of space. It uses the FIFO (first in, first out) strategy to remove the oldest items from the cache.
We’ll use flexicache with maxsize (of cache) of 2, meaning after 2 saves it starts discarding the oldest cache entries. Entries in cache functions are identified in the cache by arguments (v),so we add an argument to the function.
@flexicache(maxsize=2)
def random_func(v):
return randint(1, 1000)Let’s see how it works.
result1 = random_func(1)
# True as the function is cached
assert result1 == random_func(1)
# True as the function is cached
assert random_func(2) == random_func(2) So far so good. The cache is working as expected. Now let’s start evicting the first items added to the cache. We’ll add a third item to the cache and see if the first one is evicted.
# True as the function for 3 is cached,
# but it will evict the result of random_func2(1)
assert random_func(3) == random_func(3)
# False as the first result is no longer cached
assert result1 != random_func(1) timed_cache convenience wrapper
lru_cache is a built-in Python decorator that provides a simple way to cache the results of a function. It uses a Least Recently Used (LRU) caching strategy, which means that it keeps track of the most recently used items as based on arguments and removes the least recently used items when the cache reaches its maximum size. In other words, it takes out the latest items from the cache first when it runs out of space.
The downside is that it doesn’t have a timeout feature, so if you want to cache results for a specific amount of time, you need to implement that yourself.
fastcore.xtras.timed_cache is an implementation of flexicache that adds a timeout feature to functools.lru_cache.
from fastcore.xtras import timed_cache
# shortcut for @flexicache(time_policy(.1), maxsize=2)
@timed_cache(.1, maxsize=2)
def random_func(v):
return randint(1, 1000)
# True as the function is cached
assert random_func(1) == random_func(1)Testing the timeout is the same as before with flexicache(time_policy(.1), maxsize=2). We’ll call the function, wait for the timeout to be reached, and then call it again to see if the cache is invalidated.
# Wait long enough for the cache to expire
sleep(0.2)
# Assert False as the cache is time invalidated
assert result1 != random_func(1) Finally, confirm that the LRU cache is removing the first cached item. This is the same LRU cache set of tests we used in the section above about LRU caching. Again, we’ll add a third item to the cache and see if the first one is evicted.
result1 = random_func(1)
# True as the function is cached
assert result1 == random_func(1)
# True as the function is cached
assert random_func(2) == random_func(2)
# True as the function for 3 is cached,
# but it will evict the result of random_func2(1)
assert random_func(3) == random_func(3)
# False as the first result is no longer cached
assert result1 != random_func(1) 