A Python library providing a powerful, memory-efficient, and high-performance caching solution with tag-based invalidation. It's designed for demanding applications, such as web services built with FastAPI or Django, where quick cache access and precise invalidation are crucial. TaggedCache supports both synchronous and asynchronous functions.
- Blazing-Fast GETs: Highly optimized cache lookups using integer hash keys and minimal overhead on the hot path.
- Tag-Based Invalidation: Associate cached items with multiple string tags and invalidate groups of items by tag (e.g., "user:123", "product_category:electronics").
- Async Support: Seamlessly works with both synchronous (
def) and asynchronous (async def) functions. - Memory Efficient:
- Uses
cachetools.TTLCachefor underlying storage with configurablemaxsizeandTTL(Time-To-Live). - Internal cache keys and tag identifiers are stored as compact integer hashes (Python's
hash()for keys, 64-bitxxhashfor tags).
- Uses
- Thread-Safe: Designed for concurrent use with an internal
threading.RLockto protect shared state. - Automatic Tag Cleanup: Tags associated with items are automatically cleaned up when items are evicted due to TTL expiry or cache
maxsizelimits, preventing memory leaks from stale tag references. - Robust Tag Context: On cache misses, it intelligently resolves all function arguments (positional, keyword, defaults) using
inspect.signatureto provide a comprehensive context for dynamic tag string generation. - Handles
self/cls: Correctly managesselforclsarguments for instance/class methods in cache key generation and tag context. - Simple Decorator API: Easy to integrate using a
@cache_instance.tag(...)decorator.
pip install tagged-cachefromtagged_cacheimportTaggedCache# 1. Create cache instances with custom settingsuser_profile_cache=TaggedCache(ttl=3600, maxsize=10000) # Cache user profiles for 1 hourproduct_cache=TaggedCache(ttl=7200, maxsize=5000) # Cache products for 2 hours# 2. Define your functions and decorate themclassUserService:
@user_profile_cache.tag("user:{user_id}", "user_profile", "org:{org_id}")defget_user_profile(self, user_id: int, org_id: int):
print(f"DB HIT: Fetching profile for user {user_id} in org {org_id}")
# Simulate fetching data from a database or an external servicereturn {"id": user_id, "name": f"User {user_id}", "email": f"user{user_id}@example.com", "org_id": org_id}
classProductService:
@product_cache.tag("product:{product_id}", "product_details", "category:{category_name}")defget_product_details(self, product_id: str, category_name: str):
print(f"DB HIT: Fetching details for product {product_id} in category {category_name}")
# Simulate fetching datareturn {"id": product_id, "name": f"Product {product_id}", "price": 19.99, "category": category_name}
# 3. Use your servicesuser_service=UserService()
product_service=ProductService()
# First call - will hit the database and cache the resultprofile1=user_service.get_user_profile(user_id=101, org_id=1)
print(profile1)
product1=product_service.get_product_details(product_id="abc", category_name="electronics")
print(product1)
# Second call - will fetch from cacheprofile2=user_service.get_user_profile(user_id=101, org_id=1) # Cache HITprint(profile2)
# 4. Invalidate cache entries by tag# Imagine user 101's profile was updatedprint("\nInvalidating user:101...")
invalidated_count=user_profile_cache.invalidate_tag("user:101")
print(f"Invalidated {invalidated_count} cache entries for user:101.")
# This call will now miss the cache and re-fetchprofile_after_invalidation=user_service.get_user_profile(user_id=101, org_id=1)
print(profile_after_invalidation)
### Async Support Example```pythonimportasynciofromtagged_cacheimportTaggedCache# Create a cache instanceasync_cache=TaggedCache(ttl=3600, maxsize=1000)
# Define an async function and decorate it@async_cache.tag("user:{user_id}", "user_data")asyncdeffetch_user_data(user_id: int):
print(f"Fetching data for user {user_id}...")
# Simulate async database or API callawaitasyncio.sleep(1.0)
return {"id": user_id, "name": f"User {user_id}"}
asyncdefmain():
# First call (cache miss)user1=awaitfetch_user_data(123)
print(f"User data: {user1}")
# Second call (cache hit)user2=awaitfetch_user_data(123)
print(f"User data (cached): {user2}")
# Invalidate the taginvalidated=async_cache.invalidate_tag("user:123")
print(f"Invalidated {invalidated} entries")
# This should miss the cacheuser3=awaitfetch_user_data(123)
print(f"User data (after invalidation): {user3}")
# Run the async exampleasyncio.run(main())Constructor for creating a new cache instance.
ttl(int): Default Time-To-Live for cache entries in seconds.maxsize(int): Maximum number of entries the cache can hold. Oldest items (by TTL or LRU within TTL) are evicted when this limit is reached.
Decorator to apply to functions whose results you want to cache. Works with both sync and async functions.
tag_patterns(str): One or more f-string like patterns. The placeholders in the patterns will be filled using the decorated function's arguments (including resolved defaults) at call time.- Example:
@user_cache.tag("user:{user_id}", "role:{role_name}")
- Example:
When decorating an async def function, the decorator will return an async function that must be awaited. The original function will be awaited on cache miss, and the result will be cached.
Invalidates all cache entries associated with the exact tag_string.
tag_string(str): The specific tag to invalidate (e.g., "user:123").- Returns: The number of items actually removed from the cache.
Removes all items from this specific cache instance and clears all its tag associations.
Returns statistics about the cache.
- Returns: A dictionary with:
cache_size: Number of items in the cacheunique_tags: Number of unique tags in the cachetag_mappings: Total number of tag-to-key mappings
Returns the current number of items in the cache.
Checks if a key is in the cache. Can accept either an integer hash or a descriptive tuple.
- Cache Key Generation: For each function call, a unique key is generated based on the function's module, name, and the values of its arguments (excluding
self/cls). This descriptive key is then hashed to an integer for efficient dictionary lookups. - Tag Hashing: Tag strings (e.g., "user:123") are hashed into 64-bit integers using
xxhashfor compact storage and efficient lookup in tag-to-key mappings. - Tag-to-Key Mapping: A dictionary mapping
TagHash->Set[CacheKeyHash]. - Key-to-Tag Mapping: A dictionary mapping
CacheKeyHash->Set[TagHash]. This is crucial for cleaning up all of a key's tags when it's evicted. - Eviction Notification: A custom
TTLCachesubclass calls back intoTaggedCachewhen an item is removed, allowingTaggedCacheto clean up the associated tag mappings. - Async Function Detection: At decoration time,
inspect.iscoroutinefunction()is used to determine if a function is async, and the appropriate wrapper type (sync or async) is returned. - Conditional Await: Async functions are properly awaited when executed on cache miss, while sync functions are called directly.
# From the package root directory
pytestThis project is licensed under the MIT License - see the LICENSE file for details.