Skip to content

Caching with Redis

castlecraft-engineer provides utilities and patterns for integrating Redis as a caching layer in your applications. Caching can significantly improve performance by reducing the load on primary data stores and speeding up responses for frequently accessed data.

1. Purpose of Caching

Caching is employed in applications built with castlecraft-engineer to: * Improve Performance: Store frequently accessed data (e.g., results of expensive queries, external API responses) in a fast in-memory store like Redis to reduce latency. * Reduce Database Load: Serve data from the cache instead of hitting the database for every request, decreasing the load on your primary database. * Store Session Data or Temporary Information: Cache user session details, intermediate computation results, or other temporary data. * Rate Limiting: Implement rate limiting mechanisms by tracking request counts in Redis.

2. Configuring Redis Connections

Redis connections are configured primarily through environment variables. castlecraft-engineer supports both standard Redis connections and Redis Sentinel for high availability setups.

The core logic for establishing Redis connections resides in castlecraft_engineer.common.redis (functions like get_redis_connection and get_async_redis_connection).

Environment Variables (from common/env.py):

  • REDIS_CACHE_URL: The primary URL for connecting to a standalone Redis instance (e.g., redis://localhost:6379/0). This is used if Sentinel is not enabled.
    • Includes host, port, and optionally database number. Password can be part of the URL or set via REDIS_PASSWORD.
  • REDIS_PASSWORD: Password for the Redis instance (if REDIS_CACHE_URL doesn't include it).

  • REDIS_DB: Database number for the Redis instance (if REDIS_CACHE_URL doesn't include it, defaults to 0).

  • REDIS_SENTINEL_ENABLED: Set to "true" or "1" to enable Redis Sentinel mode.

  • REDIS_SENTINEL_HOSTS: A comma-separated list of Sentinel host:port pairs (e.g., sentinel1:26379,sentinel2:26379). Required if Sentinel is enabled.
  • REDIS_SENTINEL_MASTER_NAME: The name of the Redis master service monitored by Sentinel (e.g., mymaster). Required if Sentinel is enabled.
  • REDIS_SENTINEL_PASSWORD: Password for connecting to Sentinel nodes, if required.
  • REDIS_SENTINEL_DB: Database number to use for connections obtained via Sentinel (defaults to 0).

The library will automatically choose between standard or Sentinel setup based on REDIS_SENTINEL_ENABLED.

3. Cache Helper Functions (caching/cache.py)

The castlecraft_engineer.caching.cache module provides convenient helper functions to obtain Redis connection instances. These are typically used internally by the DI setup.

  • get_redis_cache_connection() -> redis.Redis: Returns a synchronous Redis client instance configured based on the environment variables.
  • get_redis_cache_async_connection() -> redis.asyncio.Redis: Returns an asynchronous Redis client instance (redis.asyncio.Redis) configured based on the environment variables.

4. Using Cache Clients via Dependency Injection

The recommended way to work with Redis is to obtain client instances through Dependency Injection. The ContainerBuilder from common/di.py facilitates this.

  • ContainerBuilder.with_cache(): This method registers factories for both synchronous (redis.Redis) and asynchronous (redis.asyncio.Redis) clients with the DI container.

Once registered, you can inject these clients into your services, handlers, or repositories:

import redis
import redis.asyncio as aredis # Standard alias for redis.asyncio.Redis
from castlecraft_engineer.common.di import ContainerBuilder

class MyService:
    def __init__(self, sync_cache: redis.Redis, async_cache: aredis.Redis):
        self.sync_cache = sync_cache
        self.async_cache = async_cache

    def do_sync_cache_work(self):
        self.sync_cache.set("mykey_sync", "myvalue_sync")
        value = self.sync_cache.get("mykey_sync")
        print(f"Sync cache got: {value.decode() if value else None}")

    async def do_async_cache_work(self):
        await self.async_cache.set("mykey_async", "myvalue_async")
        value = await self.async_cache.get("mykey_async")
        print(f"Async cache got: {value.decode() if value else None}")

# DI Setup
builder = ContainerBuilder()
builder.with_cache() # Registers redis.Redis and redis.asyncio.Redis factories
builder.register(MyService) # MyService can now be resolved with cache clients injected

container = builder.build()
service_instance = container.resolve(MyService)

# service_instance.do_sync_cache_work()
# asyncio.run(service_instance.do_async_cache_work()) # If running in an async context

5. Code Examples: Basic Cache Operations

Here are simple examples of using the Redis clients for common operations.

Synchronous Client (redis.Redis)

import redis
from castlecraft_engineer.common.env import REDIS_CACHE_URL # For direct connection example
from castlecraft_engineer.caching.cache import get_redis_cache_connection

# Assuming REDIS_CACHE_URL is set, e.g., "redis://localhost:6379/0"
# For DI, you'd resolve redis.Redis instance instead of calling get_redis_cache_connection directly.
try:
    sync_client: redis.Redis = get_redis_cache_connection()

    # Set a key
    sync_client.set("user:123:profile_sync", "{'name': 'Alice', 'email': 'alice@example.com'}", ex=3600) # Expires in 1 hour
    print("Sync: Set user:123:profile_sync")

    # Get a key
    user_profile_raw = sync_client.get("user:123:profile_sync")
    if user_profile_raw:
        print(f"Sync: Got user:123:profile_sync -> {user_profile_raw.decode()}")
    else:
        print("Sync: user:123:profile_sync not found.")

    # Delete a key
    sync_client.delete("user:123:profile_sync")
    print("Sync: Deleted user:123:profile_sync")

    # Check if key exists after deletion
    exists = sync_client.exists("user:123:profile_sync")
    print(f"Sync: user:123:profile_sync exists? {bool(exists)}")

    sync_client.close() # Good practice to close if not managed by DI context
except redis.exceptions.ConnectionError as e:
    print(f"Redis connection error: {e}. Ensure Redis is running and configured.")

Asynchronous Client (redis.asyncio.Redis)

import asyncio
import redis.asyncio as aredis
from castlecraft_engineer.caching.cache import get_redis_cache_async_connection

async def async_cache_operations():
    # For DI, you'd resolve redis.asyncio.Redis instance.
    try:
        async_client: aredis.Redis = get_redis_cache_async_connection()

        # Set a key
        await async_client.set("session:abc:data_async", "{'user_id': 42}", ex=600) # Expires in 10 minutes
        print("Async: Set session:abc:data_async")

        # Get a key
        session_data_raw = await async_client.get("session:abc:data_async")
        if session_data_raw:
            print(f"Async: Got session:abc:data_async -> {session_data_raw.decode()}")
        else:
            print("Async: session:abc:data_async not found.")

        # Delete a key
        await async_client.delete("session:abc:data_async")
        print("Async: Deleted session:abc:data_async")

        # Check if key exists after deletion
        exists = await async_client.exists("session:abc:data_async")
        print(f"Async: session:abc:data_async exists? {bool(exists)}")

        await async_client.close() # Good practice to close if not managed by DI context
    except redis.exceptions.ConnectionError as e:
        print(f"Async Redis connection error: {e}. Ensure Redis is running and configured.")

# To run the async example:
# if __name__ == "__main__":
#     asyncio.run(async_cache_operations())

6. Internal Usage Example: Authentication Service

A practical application of caching within castlecraft-engineer can be seen in the application/auth.py module. The AuthenticationService utilizes caching for:

  • JWKS (JSON Web Key Set) Caching: The public keys used to verify OIDC ID tokens are fetched from the OIDC provider's JWKS URI. These keys rarely change, so they are cached (typically using the asynchronous Redis client) to avoid repeated HTTP requests to the OIDC provider on every token verification.
  • Token Introspection Results: If token introspection is used, the results of introspecting a token (which involves an HTTP call to the OIDC provider's introspection endpoint) can be cached for the token's lifetime. This prevents redundant introspection calls for the same active token.

This internal usage demonstrates how caching can optimize interactions with external services and improve the overall responsiveness of authentication and authorization processes.

By properly configuring and utilizing Redis through the provided abstractions, you can significantly enhance the performance and scalability of your applications built with castlecraft-engineer.