flask cache: list keys based on a pattern?

Question

I'm using the Flask Cache plugin with Redis as backend to cache my API response. Say I have APIs to get users and create user like this:

/api/users?page=1  GET
/api/users         POST

The GET result will be cached with full URL as key. When a new user is created, I'd like to delete all keys that start with /api/users - currently I'm doing cache.clear() but it hardly seems necessary.

However, I can't seem to find an API to get a list of keys. With redis-py, there is a keys(*pattern) API for that purpose. Is there a similar API for Flask Cache?

Answer 1

Flask-Cache looks outdated. You can switch to Flask-Caching which is maintained fork of Flask-Cache.

Delete the Redis key by pattern

You have two options :

1. Connect to Redis separately using the redis package. Use core commands provided by the package to search for keys based on a pattern, and delete them.
2. Use protected methods of the Flask-Cache / Flask-Caching to get access to the underlying Redis object, and use core commands on that object to search for keys based on a pattern, and delete them.

Option 1: Separate connection to Redis

# python utility for redis
import redis

r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, password=REDIS_PASSWORD)

def delete_pattern(pattern: str) -> int:
    """Remove all keys matching pattern.
    """
    count = 0
    for key in r.scan_iter(pattern):
        r.delete(key)
        count += 1
    return count

# pass pattern to delete
CACHE_URL_PAT = "flask_cache*auth*"
delete_pattern(CACHE_URL_PAT)

Option 2: Use the protected methods to get access to the underlying Redis connection object

⚠️ Even though this works fine, but these are undocumented methods. I went through the source code on GitHub to create this function. Tested only for CACHE_TYPE = 'redis'

# initialised cache object stored in extensions file
from app.extensions import cache

def delete_pattern(pattern):
    status = False
    binary_keys = cache.cache._read_clients.keys(pattern)
    keys = [k.decode("utf-8", errors="ignore") for k in binary_keys if k]
    if keys:
        status = cache.cache._write_client.delete(*keys)
    return status

# pass pattern to delete
CACHE_URL_PAT = "flask_cache*auth*"
delete_pattern(CACHE_URL_PAT)

---

NOTE : flask_cache_ is default CACHE_KEY_PREFIX in Flask-Caching. If you have used some other value for CACHE_KEY_PREFIX, please use that (instead of flask_cache_) as the prefix for your search pattern.