Python Redis Cache

Warnings

• breaking Version 4.0.0 introduced a breaking change requiring Python 3.8 or newer. It also added support for caching functions with positional-only arguments.
  Fix: Upgrade your Python environment to 3.8+ if using version 4.0.0 or later.
• breaking Version 3.0.0 changed the key format used for storing cached items. This means cached data stored with older versions will not be accessible or compatible with versions 3.0.0 and newer.
  Fix: When upgrading from a version older than 3.0.0, consider flushing your Redis cache if you rely on existing cached data, as it will become inaccessible. New data will be stored in the new format.
• gotcha Proper configuration of the `redis-py` client is crucial. Common issues include incorrect host/port, authentication failures, or not setting `decode_responses=True` if you expect string outputs directly from Redis, which can lead to byte-string issues in your application logic.
  Fix: Always test your `redis.StrictRedis` or `redis.Redis` connection independently (e.g., with `client.ping()`) before passing it to `RedisCache`. Pay attention to `host`, `port`, `db`, `password`, and `decode_responses` parameters.
• gotcha Caching class or instance methods requires careful handling. By default, `self` (the instance itself) is part of the cache key, meaning each instance will have its own cache. When invalidating, you must provide the instance to `delete_memoized` if `self` was part of the key.
  Fix: Be explicit about how `self` affects your cache keys. If you want a class method to be cached globally regardless of instance, you might need to adjust key generation or consider caching outside the method. For instance methods, use `cache.delete_memoized(YourClass.method, obj_instance=your_instance)` for invalidation.

Install

• pip install python-redis-cache Install stable version

Imports

• RedisCache

  from redis_cache import RedisCache

Quickstart

This quickstart demonstrates how to initialize `RedisCache` with a `redis-py` client and use the `@cache` decorator on a function. It shows how the cache stores results for subsequent calls with the same arguments, and how to manually invalidate a specific cache entry. Ensure a Redis server is accessible at the specified host and port (or defaults).

import os
import redis
from redis_cache import RedisCache
import time

# Ensure Redis server is running (e.g., docker run --name some-redis -p 6379:6379 -d redis)
# Get Redis connection details from environment variables, or use defaults
REDIS_HOST = os.environ.get('REDIS_HOST', 'localhost')
REDIS_PORT = int(os.environ.get('REDIS_PORT', '6379'))
REDIS_DB = int(os.environ.get('REDIS_DB', '0'))

try:
    # Connect to Redis. decode_responses=True is often useful.
    redis_client = redis.StrictRedis(
        host=REDIS_HOST,
        port=REDIS_PORT,
        db=REDIS_DB,
        decode_responses=True
    )
    redis_client.ping() # Test connection
    print(f"Successfully connected to Redis at {REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}")

    # Initialize RedisCache with the Redis client and a TTL (Time To Live)
    cache = RedisCache(redis_client=redis_client, ttl=30) # Cache entries for 30 seconds

    @cache
    def get_user_data(user_id: int):
        """Simulates fetching user data from a slow external source."""
        print(f"Fetching user {user_id} from original source...")
        time.sleep(1) # Simulate network delay
        return {"id": user_id, "name": f"User {user_id}", "fetch_time": time.time()}

    print("\n--- First call (should fetch from source) ---")
    user_data1 = get_user_data(101)
    print(f"User Data 1: {user_data1}")

    print("\n--- Second call (should be served from cache) ---")
    user_data2 = get_user_data(101)
    print(f"User Data 2: {user_data2}") # fetch_time should be the same as user_data1

    print("\n--- Third call (different user, fetches from source) ---")
    user_data3 = get_user_data(102)
    print(f"User Data 3: {user_data3}")

    print("\n--- Invalidate cache for user 101 and re-fetch ---")
    cache.delete_memoized(get_user_data, 101) # Invalidate specific cache entry
    user_data4 = get_user_data(101) # Should fetch from source again
    print(f"User Data 4: {user_data4}")

except redis.exceptions.ConnectionError as e:
    print(f"ERROR: Could not connect to Redis at {REDIS_HOST}:{REDIS_PORT}. "
          "Please ensure Redis server is running. Quickstart cannot proceed.")
except Exception as e:
    print(f"An unexpected error occurred: {e}")