Caching tool for python, working with Redis single instance and Redis cluster mode

PyPi link

Features

• ✅ Automatic reconnection: Redis client handles reconnection automatically
• ✅ Graceful degradation: Falls back to in-memory cache when Redis is unavailable
• ✅ No crashes: All operations handle errors gracefully
• ✅ Async & Sync support: Works with both async and sync functions
• ✅ Redis Cluster support: Works with single Redis instance and Redis Cluster
• ✅ Custom encoders/decoders: Support for custom serialization

Installation

 $ pip install cache-house 

or with poetry

poetry add cache-house

---

Quick Start

---

Cache decorator works with both async and sync functions. The library automatically handles Redis reconnection and falls back to in-memory cache when Redis is unavailable.

fromcache_house.backendsimportRedisFactoryfromcache_house.cacheimportcacheimportasyncio# Initialize Redis with fallback enabled (recommended for production)RedisFactory.init(fallback_to_memory=True)
@cache() # default expire time is 180 secondsasyncdeftest_cache(a: int, b: int):
print("async cached - this only prints on cache miss")
return [a, b]
@cache()deftest_cache_1(a: int, b: int):
print("cached - this only prints on cache miss")
return [a, b]
if__name__=="__main__":
print("First call (cache miss):")
print(test_cache_1(3, 4))
print("\nSecond call (cache hit):")
print(test_cache_1(3, 4)) # This will use cache, print won't executeprint("\nAsync function:")
print(asyncio.run(test_cache(1, 2)))
print(asyncio.run(test_cache(1, 2))) # Cached

Output:

First call (cache miss):
cached - this only prints on cache miss
[3, 4]
Second call (cache hit):
[3, 4] # No print - served from cache
Async function:
async cached - this only prints on cache miss
[1, 2]
[1, 2] # Cached

Check stored cache keys:

➜ $ rdcli KEYS "*"
1) cachehouse:main:8f65aed1010f0062a783c83eb430aca0
2) cachehouse:main:f665833ea64e4fc32653df794257ca06

---

Setup Redis cache instance

---

You can pass all redis-py arguments to RedisFactory.init method and additional arguments:

defRedisFactory.init(
host: str="localhost",
port: int=6379,
encoder: Callable[..., Any] = ...,
decoder: Callable[..., Any] = ...,
namespace: str= ...,
key_prefix: str= ...,
key_builder: Callable[..., Any] = ...,
password: str= ...,
db: int= ...,
cluster_mode: bool=False, # Force cluster mode (skip auto-detection)autodetect_cluster: bool=True, # Auto-detect if Redis is running in cluster modefallback_to_memory: bool=True, # Enable in-memory fallback when Redis is unavailable**redis_kwargs
)

Cluster auto-detection

By default (autodetect_cluster=True), RedisFactory.init will:

• Try to send CLUSTER INFO to the target Redis node
• If the command succeeds → cluster mode is detected, and RedisClusterCache is used internally
• If the command fails with a Redis error → standalone mode is assumed, and RedisCache is used

This means you can usually just call:

fromcache_house.backendsimportRedisFactoryRedisFactory.init(
host="localhost",
port=6379,
fallback_to_memory=True,
# autodetect_cluster=True by default
)

and cache-house will automatically choose the correct backend (standalone or cluster) based on the Redis server configuration.

Explicit modes (optional)

• Force standalone Redis (no detection):

RedisFactory.init(
host="localhost",
port=6379,
cluster_mode=False,
autodetect_cluster=False, # Always use standalone RedisCache
)

• Force Redis Cluster (no detection):

RedisFactory.init(
host="localhost",
port=6379,
cluster_mode=True, # Always use RedisClusterCacheautodetect_cluster=False, # Optional, explicit
)

Best Practice: Initialize Redis with fallback enabled

fromcache_house.backendsimportRedisFactory# Initialize with fallback to memory cache (default: True)# Your application will continue working even if Redis is temporarily unavailableRedisFactory.init(
host="localhost",
port=6379,
password="your_password", # Optionaldb=0,
fallback_to_memory=True# Falls back to in-memory cache when Redis is down
)

Custom encoder and decoder

fromcache_house.backendsimportRedisFactoryimportjsondefcustom_encoder(data):
returnjson.dumps(data)
defcustom_decoder(data):
returnjson.loads(data)
RedisFactory.init(
encoder=custom_encoder, decoder=custom_decoder,
fallback_to_memory=True
)

Default encoder and decoder is pickle module.

---

Setup Redis Cluster cache instance

---

All manipulation with RedisCache is the same with RedisClusterCache

fromcache_house.backendsimportRedisFactoryfromcache_house.cacheimportcache# Initialize Redis Cluster with fallback enabledRedisFactory.init(
cluster_mode=True,
startup_nodes=[
{"host": "127.0.0.1", "port": "7000"},
{"host": "127.0.0.1", "port": "7001"},
],
fallback_to_memory=True# Falls back to in-memory cache when cluster is unavailable
)
@cache()asyncdeftest_cache(a: int, b: int):
print("cached")
return [a, b]

Redis Cluster parameters (all redis-py cluster arguments are supported):

RedisFactory.init(
cluster_mode=True,
startup_nodes=[{"host": "127.0.0.1", "port": "7000"}],
cluster_error_retry_attempts: int=3,
require_full_coverage: bool=True,
skip_full_coverage_check: bool=False,
reinitialize_steps: int=10,
read_from_replicas: bool=False,
fallback_to_memory: bool=True,
**redis_kwargs
)

---

Setup cache instance with FastAPI

---

Best Practice: Initialize Redis in startup event with fallback enabled. Your application will continue working even if Redis is temporarily unavailable.

importloggingimportuvicornfromfastapi.applicationsimportFastAPIfromcache_house.backendsimportRedisFactoryfromcache_house.cacheimportcacheapp=FastAPI()
@app.on_event("startup")asyncdefstartup():
# Initialize with fallback - app won't crash if Redis is unavailableRedisFactory.init(
host="localhost",
port=6379,
fallback_to_memory=True# Enable in-memory fallback
)
print("App started - Redis cache initialized")
@app.on_event("shutdown")asyncdefshutdown():
# Gracefully close connectionsRedisFactory.close_connections()
print("App shutdown - Redis connections closed")
@app.get("/notcached")asyncdeftest_route():
print("notcached")
return {"hello": "world"}
@app.get("/cached")@cache(expire=60) # Cache for 60 secondsasyncdeftest_route():
print("cached") # This print only runs on cache missreturn {"hello": "world"}
@app.get("/cached-with-custom-expire")@cache(expire=300, namespace="api") # Cache for 5 minutes with custom namespaceasyncdefexpensive_operation():
# Simulate expensive operationimporttimetime.sleep(1)
return {"result": "expensive computation"}
if__name__=="__main__":
uvicorn.run(app, port=8033)

---

Cache decorator options

---

You can set expire time (seconds or timedelta), namespace, and key prefix in the cache decorator:

fromdatetimeimporttimedeltafromcache_house.cacheimportcache# Using seconds@cache(expire=30, namespace="app", key_prefix="test")asyncdeftest_cache(a: int, b: int):
print("cached")
return [a, b]
# Using timedelta@cache(expire=timedelta(minutes=5), namespace="app", key_prefix="test")deftest_cache_sync(a: int, b: int):
print("cached")
return [a, b]
if__name__=="__main__":
print(asyncio.run(test_cache(1, 2)))
print(test_cache_sync(3, 4))

Check stored cache:

rdcli KEYS "*"
1) test:app:f665833ea64e4fc32653df794257ca06

---

Understanding Namespaces and Key Builders

---

Namespaces

Namespaces help organize your cache keys and make it easier to manage different parts of your application. The default namespace is "main".

Key Format: {prefix}:{namespace}:{hash}

Example with different namespaces:

fromcache_house.backendsimportRedisFactoryfromcache_house.cacheimportcacheRedisFactory.init(fallback_to_memory=True)
# API endpoints namespace@cache(expire=60, namespace="api")defget_user(user_id: int):
return {"id": user_id, "name": f"User {user_id}"}
# Database queries namespace@cache(expire=300, namespace="database")defget_user_posts(user_id: int):
return [{"id": 1, "title": "Post 1"}]
# Configuration namespace@cache(expire=3600, namespace="config")defget_app_config():
return {"setting": "value"}
# Default namespace (if not specified)@cache(expire=180)defdefault_function():
return"default"

Cache keys will be:

cachehouse:api:abc123...
cachehouse:database:def456...
cachehouse:config:ghi789...
cachehouse:main:jkl012... # default namespace

Benefits of using namespaces:

• Organization: Group related cache entries together
• Easy cleanup: Clear all keys in a specific namespace
• Multi-tenancy: Separate cache for different applications/services
• Debugging: Easier to identify cache keys in Redis

Global Namespace Configuration

You can set a default namespace for all cache operations:

fromcache_house.backendsimportRedisFactory# Set default namespace for all cache operationsRedisFactory.init(
namespace="myapp", # All cache keys will use "myapp" namespace by defaultkey_prefix="app", # Change default prefix from "cachehouse" to "app"fallback_to_memory=True
)
# This will use "myapp" namespace@cache(expire=60)defmy_function():
return"data"# Override namespace for specific function@cache(expire=60, namespace="special")defspecial_function():
return"special data"

Resulting keys:

app:myapp:abc123... # default namespace
app:special:def456... # overridden namespace

Key Prefix

The key prefix is the first part of every cache key. Default is "cachehouse".

# Global prefixRedisFactory.init(key_prefix="myapp", namespace="v1")
# Per-decorator prefix (overrides global)@cache(expire=60, key_prefix="api", namespace="users")defget_user(id: int):
return {"id": id}

Key format: {key_prefix}:{namespace}:{hash}

Custom Key Builder

You can create your own key builder function for complete control over cache key generation:

importhashlibfromcache_house.backendsimportRedisFactoryfromcache_house.cacheimportcachedefcustom_key_builder(module, name, args, kwargs, prefix="cachehouse", namespace="main"):
""" Custom key builder function Args: module: Function's module name name: Function name args: Function positional arguments kwargs: Function keyword arguments prefix: Key prefix namespace: Namespace """# Example: Create a more readable key# Format: prefix:namespace:module.function:arg1:arg2:kwarg1=value1key_parts= [prefix, namespace, f"{module}.{name}"]
# Add positional argumentsforarginargs:
key_parts.append(str(arg))
# Add keyword argumentsfork, vinsorted(kwargs.items()):
key_parts.append(f"{k}={v}")
# Join and create hash for long keyskey_string=":".join(key_parts)
iflen(key_string) >200: # Redis key length limitkey_hash=hashlib.md5(key_string.encode()).hexdigest()
returnf"{prefix}:{namespace}:{key_hash}"returnkey_string# Use custom key builder globallyRedisFactory.init(
key_builder=custom_key_builder,
namespace="custom",
fallback_to_memory=True
)
@cache(expire=60)defmy_function(a: int, b: int, name: str="test"):
return {"result": a+b, "name": name}

Or use custom key builder per decorator:

defsimple_key_builder(module, name, args, kwargs, prefix="cache", namespace="app"):
# Simple key: just use function name and first argumentfirst_arg=args[0] ifargselse"default"returnf"{prefix}:{namespace}:{name}:{first_arg}"@cache(expire=60, key_builder=simple_key_builder, namespace="simple")defget_item(item_id: int):
return {"id": item_id}

Key builder function signature:

defkey_builder(
module: str, # Function's module (e.g., "__main__" or "myapp.services")name: str, # Function nameargs: tuple, # Positional argumentskwargs: dict, # Keyword argumentsprefix: str, # Key prefixnamespace: str# Namespace
) ->str:
# Return the cache key as a stringreturn"your:custom:key:format"

Clearing Cache by Namespace

You can clear all cache keys in a specific namespace:

fromcache_house.backendsimportRedisCache# Clear all keys in a namespaceRedisCache.clear_keys("cachehouse:api") # Clears all keys starting with "cachehouse:api"# Or with custom prefixRedisCache.clear_keys("myapp:database") # Clears all keys in "database" namespace

Example: Clear cache for a specific namespace

fromcache_house.backendsimportRedisFactory, RedisCacheRedisFactory.init(namespace="myapp", fallback_to_memory=True)
# Cache some data@cache(expire=300, namespace="users")defget_user(id: int):
return {"id": id}
@cache(expire=300, namespace="posts")defget_post(id: int):
return {"id": id}
# Later, clear only "users" namespaceRedisCache.clear_keys("cachehouse:users") # Only clears users cache# Posts cache remains intact

Best Practices for Namespaces

1. Use descriptive namespaces:

   @cache(namespace="api.users") # Good@cache(namespace="x") # Bad - not descriptive

2. Organize by feature or service:

   namespace="api.users"namespace="api.products"namespace="database.queries"namespace="external.api"

3. Use consistent naming:

   # Good - consistent patternnamespace="v1.api"namespace="v1.database"namespace="v2.api"

4. Set global namespace for multi-tenant apps:

   # Different namespace per tenanttenant_id=get_current_tenant()
   RedisFactory.init(namespace=f"tenant_{tenant_id}")

5. Use namespaces for cache invalidation:

   # When user data changes, clear user namespacedefupdate_user(user_id):
   # ... update logic ...RedisCache.clear_keys("cachehouse:users") # Clear all user cache

---

Custom encoder and decoder in decorator

---

If your function works with non-standard data types, you can pass custom encoder and decoder functions to the cache decorator:

importasyncioimportjsonfromcache_house.backendsimportRedisFactoryfromcache_house.cacheimportcacheRedisFactory.init(fallback_to_memory=True)
defcustom_encoder(data):
returnjson.dumps(data)
defcustom_decoder(data):
returnjson.loads(data)
@cache(expire=30, encoder=custom_encoder, decoder=custom_decoder, namespace="custom")asyncdeftest_cache(a: int, b: int):
print("async cached")
return {"a": a, "b": b}
@cache(expire=30)deftest_cache_1(a: int, b: int):
print("cached")
return [a, b]
if__name__=="__main__":
print(asyncio.run(test_cache(1, 2)))
print(test_cache_1(3, 4))

Check stored cache:

rdcli KEYS "*"
1) cachehouse:main:8f65aed1010f0062a783c83eb430aca0
2) cachehouse:custom:f665833ea64e4fc32653df794257ca06

---

Error Handling and Resilience

---

cache-house is designed to be resilient and won't crash your application:

Automatic Reconnection

Redis client handles reconnection automatically. You don't need to manage connections manually.

In-Memory Fallback

When Redis is unavailable, cache operations automatically fall back to in-memory cache:

fromcache_house.backendsimportRedisFactoryfromcache_house.cacheimportcache# Initialize with fallback enabled (default: True)RedisFactory.init(
host="localhost",
port=6379,
fallback_to_memory=True# Falls back to in-memory cache when Redis is down
)
@cache(expire=60)defexpensive_operation(data):
# This function will work even if Redis is unavailable# Results will be cached in memory temporarilyreturnprocess_data(data)

Graceful Error Handling

All cache operations handle errors gracefully:

fromcache_house.backendsimportRedisFactoryfromcache_house.cacheimportcache# Even if Redis is not initialized, your code won't crashcache_instance=RedisFactory.get_instance()
ifcache_instanceisNone:
print("Cache not available, but app continues running")
@cache(expire=60)defmy_function():
# If Redis fails, this function still executes normally# Cache errors are logged but don't crash the appreturnexpensive_computation()

Best Practices

1. Always enable fallback for production:

   RedisFactory.init(fallback_to_memory=True)

2. Handle cache as optional:

   @cache(expire=60)defmy_function():
   # Function works with or without cachereturncompute_result()

3. Use appropriate expiration times:

   @cache(expire=300) # 5 minutes for stable datadefget_stable_data():
   returnfetch_data()
   @cache(expire=30) # 30 seconds for frequently changing datadefget_dynamic_data():
   returnfetch_data()

4. Close connections on shutdown (e.g., in FastAPI):

   @app.on_event("shutdown")asyncdefshutdown():
   RedisFactory.close_connections()

---

Complete Example: Production-Ready Setup

---

Here's a complete example showing best practices for using cache-house in production:

importasyncioimportloggingfromdatetimeimporttimedeltafromcache_house.backendsimportRedisFactoryfromcache_house.cacheimportcache# Configure logging to see cache operationslogging.basicConfig(level=logging.INFO)
# Initialize Redis with fallback enabled# Your app will work even if Redis is temporarily unavailableRedisFactory.init(
host="localhost",
port=6379,
password=None, # Set if your Redis requires authenticationdb=0,
fallback_to_memory=True, # Enable in-memory fallback# You can pass any redis-py connection arguments heresocket_connect_timeout=5,
socket_timeout=5,
)
# Example 1: Cache expensive computation@cache(expire=300) # Cache for 5 minutesdefexpensive_computation(n: int):
"""This expensive operation will be cached"""result=sum(i*iforiinrange(n))
print(f"Computed result for {n}: {result}")
returnresult# Example 2: Cache API response@cache(expire=60, namespace="api") # Cache for 1 minute with namespaceasyncdeffetch_user_data(user_id: int):
"""Simulate API call - will be cached"""print(f"Fetching user {user_id} from API...")
awaitasyncio.sleep(0.1) # Simulate network delayreturn {"user_id": user_id, "name": f"User {user_id}"}
# Example 3: Cache with custom expiration@cache(expire=timedelta(hours=1), namespace="long_term")defget_configuration():
"""Configuration that changes rarely"""print("Loading configuration...")
return {"setting1": "value1", "setting2": "value2"}
# Example 4: Cache database query result@cache(expire=180, namespace="database")asyncdefget_user_posts(user_id: int):
"""Simulate database query"""print(f"Querying database for user {user_id} posts...")
awaitasyncio.sleep(0.05)
return [{"id": 1, "title": "Post 1"}, {"id": 2, "title": "Post 2"}]
asyncdefmain():
print("=== Example 1: Expensive computation ===")
print(expensive_computation(1000000)) # First call - computesprint(expensive_computation(1000000)) # Second call - from cacheprint("\n=== Example 2: API response caching ===")
print(awaitfetch_user_data(1)) # First call - fetchesprint(awaitfetch_user_data(1)) # Second call - from cacheprint("\n=== Example 3: Configuration caching ===")
print(get_configuration()) # First call - loadsprint(get_configuration()) # Second call - from cacheprint("\n=== Example 4: Database query caching ===")
print(awaitget_user_posts(1)) # First call - queriesprint(awaitget_user_posts(1)) # Second call - from cache# Clean upRedisFactory.close_connections()
if__name__=="__main__":
asyncio.run(main())

Output:

INFO:cache_house.backends.redis_backend:redis initialized (Redis will handle reconnections automatically)
=== Example 1: Expensive computation ===
Computed result for 1000000: 333333333333500000
Computed result for 1000000: 333333333333500000
=== Example 2: API response caching ===
Fetching user 1 from API...
{'user_id': 1, 'name': 'User 1'}
{'user_id': 1, 'name': 'User 1'}
...

Note: If Redis is unavailable, all operations will still work using the in-memory fallback cache. Your application won't crash!

---

All examples work with both Redis Cluster and single Redis instance.

---

Contributing

Free to open issue and send PR

cache-house supports Python >= 3.10