RedisManager¶

The RedisManager class handles Redis connections and atomic operations with automatic retries.

Class Definition¶

class RedisManager:
    """
    Robust Redis handler with connection pooling and automatic reconnection.
    """

Key Methods¶

initialize¶

def initialize(self):
    """Initialize Redis connection with retry logic"""

get_connection¶

@contextmanager
def get_connection(self):
    """Context manager for safe Redis operations"""

safe_operation¶

def safe_operation(self, func, *args, **kwargs):
    """Execute Redis operation with error handling"""

Atomic Operations¶

get_key¶

def get_key(self, namespace: str, key: str) -> Any:
    """Get namespaced key with prefix"""

set_key¶

def set_key(self, namespace: str, key: str, value: Any, ttl: int | None = None) -> bool:
    """Set namespaced key with optional TTL"""

incr¶

def incr(self, namespace: str, key: str, ttl: int | None = None) -> int:
    """Atomic increment with expiration"""

Usage Example¶

from flaskapi_guard.handlers.redis_handler import RedisManager
from flaskapi_guard.models import SecurityConfig

config = SecurityConfig(redis_url="redis://localhost:6379")
redis = RedisManager(config)

def example():
    redis.initialize()
    with redis.get_connection() as conn:
        conn.set("test_key", "value")

    # Atomic operation
    redis.set_key("namespace", "key", "value", ttl=3600)