Caching¶

The caching module provides a sophisticated two-layer caching system built on top of Flask-Caching, designed for high-performance applications with both local RAM caching and shared Redis caching capabilities.

Overview¶

The caching system addresses the need for efficient data caching across distributed applications by providing:

Two-layer architecture: Local RAM cache for ultra-fast access + shared Redis cache for persistence and cross-instance sharing
Intelligent cache management: Automatic key tracking, bulk deletion, and cache statistics
Async computation: Background cache building with worker jobs to prevent blocking
Request-scoped caching: Automatic cleanup at request end
Atomic operations: Race condition prevention for cache writes

Quick Start¶

Basic Function Caching¶

The simplest way to cache a function's return value:

from shared.caching.cache import cached_for

@cached_for(minutes=30)
def get_user_profile(user_id: int):
    # Expensive database query or API call
    return fetch_user_from_database(user_id)

# First call: executes function and caches result
profile = get_user_profile(123)

# Subsequent calls within 30 minutes: returns cached result
profile = get_user_profile(123)  # Fast cache hit

Request-Scoped Caching¶

For caching within a single HTTP request:

from shared.caching.cache import request_cached

@request_cached()
def get_current_user_permissions():
    # This will be cached only for the duration of the current request
    return expensive_permission_calculation()

Manual Cache Management¶

For direct cache operations:

from shared.caching.cache import alan_cache
from datetime import timedelta

# Set a value
alan_cache.set("my_key", {"some": "value"}, timedelta(hours=1))

# Get a value
data = alan_cache.get("my_key")

# Check if key exists
if alan_cache.has("my_key"):
    print("Key exists in cache")

# Delete a key
alan_cache.delete("my_key")

Core Concepts¶

Two-Layer Architecture¶

The caching system uses a dual-layer approach:

Local RAM Cache (SimpleCache)
Ultra-fast in-memory storage
Process-specific
First layer checked on cache hits
Automatically expires with process restart
Shared Redis Cache
Persistent across process restarts
Shared between application instances
Second layer for cache hits
Supports advanced features like atomic writes

@cached_for(minutes=60)
def expensive_computation(param):
    # Cache lookup order:
    # 1. Local RAM cache (fastest)
    # 2. Redis cache (shared)
    # 3. Execute function if not found
    return perform_calculation(param)

Cache Keys and Invalidation¶

Cache keys are automatically generated from function names and parameters:

@cached_for(hours=2, cache_key_with_full_args=True)
def get_order_summary(user_id: int, order_id: int):
    return calculate_order_summary(user_id, order_id)

# Invalidate specific cache entry
alan_cache.clear_cached_func(get_order_summary, 123, 456)

# Invalidate all cache entries for this function
alan_cache.clear_cached_func_all(get_order_summary)

# Partial invalidation (requires cache_key_with_full_args=True)
count, is_running = alan_cache.clear_cached_func_some(
    get_order_summary, 
    123  # Clear all orders for user_id=123
)

Common Usage Patterns¶

1. Basic Function Caching with Expiration¶

from shared.caching.cache import cached_for

@cached_for(hours=1)
def get_product_catalog():
    """Cache product catalog for 1 hour."""
    return fetch_products_from_api()

@cached_for(minutes=15)
def get_user_dashboard_data(user_id: int):
    """Cache user-specific dashboard data for 15 minutes."""
    return build_dashboard_data(user_id)

2. Cache-Only-If Conditional Caching¶

@cached_for(
    minutes=30,
    unless=lambda func, user_id, *args, **kwargs: user_id is None
)
def get_user_preferences(user_id: int | None):
    """Cache user preferences, but not for anonymous users."""
    if user_id is None:
        return get_default_preferences()
    return fetch_user_preferences(user_id)

3. Local RAM Cache Only (No Redis)¶

@cached_for(minutes=5, local_ram_cache_only=True)
def get_temporary_session_data(session_id: str):
    """Cache in RAM only for security-sensitive session data."""
    return fetch_session_data(session_id)

4. Redis Cache Only (No RAM)¶

@cached_for(hours=2, shared_redis_cache_only=True)
def get_large_dataset(query_params: dict):
    """Cache large datasets in Redis only to avoid memory pressure on local instances."""
    return fetch_large_data_from_database(query_params)

@cached_for(days=1, shared_redis_cache_only=True)
def get_configuration_data():
    """Cache configuration in Redis to share across all application instances."""
    return load_app_configuration()

5. Async Cache Computation¶

For expensive operations that shouldn't block the main thread:

from shared.helpers.asynchronous.exceptions import (
    AsyncValueBeingBuiltException, 
    AsyncValueFailedToBeBuiltException
)

@cached_for(
    hours=2,
    async_compute=True,
    async_compute_job_timeout=timedelta(minutes=5)
)
def generate_monthly_report(company_id: int, month: str):
    """Generate report asynchronously to avoid blocking web requests."""
    return create_complex_report(company_id, month)

# Handle different async cache states
try:
    report = generate_monthly_report(123, "2024-01")
    # Success: cached result available or just computed
    return {"status": "success", "data": report}
except AsyncValueBeingBuiltException:
    # Report is being generated in background
    return {"status": "generating", "message": "Report will be ready soon"}
except AsyncValueFailedToBeBuiltException:
    # Background computation failed - handle gracefully
    return {"status": "error", "message": "Report generation failed, please try again"}

6. Periodic Cache Refresh¶

@cached_for(
    hours=6,
    async_refresh_every=timedelta(hours=4),
    warmup_on_startup=True
)
def get_global_configuration():
    """Refresh configuration every 4 hours, cache for 6 hours."""
    return fetch_config_from_external_service()

7. Request-Scoped Caching with HTTP Method Control¶

@request_cached(for_http_methods={"GET", "POST"})
def get_user_context():
    """Cache user context for the current request only."""
    return build_user_context_from_session()

8. Object Instance Caching with Automatic Cleanup¶

@cached_for(
    minutes=30, 
    expire_when="object_is_destroyed",
)
def get_computed_property(self):
    """Cache expensive property calculation until object is destroyed."""
    return self.perform_expensive_calculation()

9. Cache with Custom Key Prefix¶

@cached_for(
    hours=1,
    cache_key_prefix=lambda func, version, *args, **kwargs: f"v{version}"
)
def get_api_data(version: str, endpoint: str):
    """Cache with version-specific key prefix for API versioning."""
    return fetch_from_api(version, endpoint)

10. Atomic Cache Writes (Race Condition Prevention)¶

Atomic writes prevent race conditions when multiple processes try to cache the same value simultaneously. This is especially useful if the function has a side effect that you want performed only once (like send a message to slack when the function is computed)

at_least_once: Guarantees the function runs at least once, but may run multiple times if there are race conditions. Use when duplicate computation is acceptable but missing computation is not.
at_most_once: Guarantees the function runs at most once across all processes. Only one process will execute the function, others will wait for the result. Use for expensive singleton initialization where duplicate execution must be avoided.

# Use at_most_once for expensive singleton initialization
@cached_for(
    hours=1,
    atomic_writes="at_most_once"
)
def initialize_expensive_resource():
    """Ensure only one process initializes this resource."""
    return create_expensive_singleton_resource()

# Use at_least_once when some duplication is acceptable
@cached_for(
    minutes=30,
    atomic_writes="at_least_once"
)
def fetch_external_data(api_endpoint: str):
    """Prevent most race conditions while allowing some duplicate API calls."""
    return call_external_api(api_endpoint)

Performance Considerations¶

Cache Key Design Patterns¶

Optimize cache performance through smart key design.

from shared.caching.cache import cached_for
from hashlib import md5

# ❌ BAD: Keys count is very large (because of user_id), keys size can grow very large
@cached_for(hours=1)
def bad_search_results(user_id: int, query: str, filters: dict):
    # Cache key includes user_id, entire filter...
    return perform_search(query, filters)

# ✅ GOOD: ignore high cardinality args (user_id) not useful for caching. Keys size can be quite large still
@cached_for(
    hours=1,
    args_to_ignore=['user_id'],
)
def good_search_results(query: str, filters: dict):
    return perform_search(query, filters)

# ✅ BETTER: Use specific cache keys with ignored parameters

def _encode(*args):
    return str(md5(str("|".join(args)).encode()).hexdigest()

@cached_for(
    hours=1,
    cache_key_with_func_args = False, # Don't include any arg by default
    # craft our cache key ourselves
    cache_key_prefix=lambda func, query, filters: _encode(query, filters.items())
)
def shared_search_results(query: str, user_id: int, filters: dict | None = None):
    """Search results shared across users, personalization handled separately."""
    base_results = perform_search(query, filters or {})
    # Apply user-specific personalization without affecting cache
    return personalize_results(base_results, user_id)

Memory Management¶

Local cache has memory limits (CACHE_THRESHOLD)
Use local_ram_cache_only=True for temporary data
Monitor cache hit rates with built-in metrics

Reference¶

API Reference - Complete API documentation for all caching classes and functions
Cache Backends - Detailed documentation of Redis, SimpleCache, and custom backend implementations