Redis Integration

Redis Integration

Redis integration for sillo web framework, providing seamless caching, session storage, and data management capabilities with comprehensive dependency injection support.

Features

• High-Performance Redis Client: Async Redis client with connection pooling and health checks
• Dependency Injection: Simplified DI pattern for clean, testable code
• Connection Management: Automatic connection lifecycle management
• Data Structure Support: Strings, Hashes, Lists, Sets, JSON operations
• Configuration: Environment-based and programmatic configuration
• Error Handling: Robust error handling with custom exceptions
• Testing Support: Easy mocking and testing utilities
• Monitoring: Built-in connection health monitoring

Installation

# Install the Redis contrib package
uv add "sillo-contrib-redis[redis]"

# Or install from the contrib meta package
uv add sillo-contrib[redis]

Requirements

• Python 3.10+
• sillo 0.0.1a1+
• Redis server (4.0+ recommended)
• redis-py (auto-installed)

Quick Start

Basic Setup

from sillo import silloApp
from sillo_contrib.redis import init_redis

app = silloApp()

# Initialize Redis with default settings
init_redis(app)

@app.get("/cache/{key}")
async def get_cache(request: Request, response: Response, key: str):
    from sillo_contrib.redis import redis_get

    value = await redis_get(key)
    if value is None:
        return response.json({"error": "Key not found"}, status_code=404)

    return {"key": key, "value": value}

With Custom Configuration

from sillo import silloApp
from sillo_contrib.redis import init_redis, RedisConfig

app = silloApp()

# Custom Redis configuration
config = RedisConfig(
    url="redis://localhost:6379/1",
    password="your_password",
    db=1,
    decode_responses=True
)

# Initialize with custom config
init_redis(app, config=config)

Environment Configuration

from sillo_contrib.redis import RedisConfig

# Load configuration from environment variables
config = RedisConfig.from_env()  # Looks for REDIS_* variables

# Or with custom prefix
config = RedisConfig.from_env("MY_REDIS_")

Dependency Injection

Basic Pattern

from sillo import silloApp, Depend
from sillo_contrib.redis import get_redis

app = silloApp()

@app.get("/user/{user_id}")
async def get_user(
    request: Request,
    response: Response,
    user_id: str,
    redis: Depend(get_redis)  # Inject RedisClient
):
    # Use redis client directly
    cached_user = await redis.json_get(f"user:{user_id}")
    if cached_user:
        return {"user": cached_user, "source": "cache"}

    return {"user": {"id": user_id, "name": "Unknown"}, "source": "default"}

Simplified Pattern

from sillo import silloApp
from sillo_contrib.redis import RedisDepend

app = silloApp()

@app.get("/user/{user_id}")
async def get_user(
    request: Request,
    response: Response,
    user_id: str,
    redis: RedisClient = RedisDepend()  # Cleaner syntax
):
    user_data = await redis.json_get(f"user:{user_id}")
    return {"user": user_data or {"id": user_id, "name": "Unknown"}}

Redis Operations

String Operations

from sillo_contrib.redis.utils import (
    redis_get, redis_set, redis_delete, redis_exists,
    redis_expire, redis_ttl, redis_incr, redis_decr
)

# Get and set strings
await redis_set("user:123:name", "John Doe", ex=3600)  # Expire in 1 hour
name = await redis_get("user:123:name")

# Atomic increment
count = await redis_incr("counter:page_views", 1)

# Check existence
exists = await redis_exists("user:123:name")  # Returns 1 if exists

# Delete keys
deleted = await redis_delete("user:123:name", "user:123:email")

Hash Operations

from sillo_contrib.redis.utils import redis_hget, redis_hset, redis_hgetall

# Set hash fields
await redis_hset("user:123", "name", "John Doe")
await redis_hset("user:123", "email", "john@example.com")

# Get single field
name = await redis_hget("user:123", "name")

# Get all fields
user_data = await redis_hgetall("user:123")
# Returns: {"name": "John Doe", "email": "john@example.com"}

List Operations

from sillo_contrib.redis import redis_lpush, redis_rpush, redis_lpop, redis_llen

# Add to list
await redis_lpush("messages", "Hello", "World")
await redis_rpush("messages", "!")

# Get list length
length = await redis_llen("messages")  # Returns 3

# Pop from list
first_msg = await redis_lpop("messages")  # Returns "World"
last_msg = await redis_rpop("messages")   # Returns "!"

Set Operations

from sillo_contrib.redis import redis_sadd, redis_smembers, redis_srem, redis_scard

# Add to set
added = await redis_sadd("tags", "python", "redis", "cache")

# Get all members
tags = await redis_smembers("tags")
# Returns: ["python", "redis", "cache"]

# Remove members
removed = await redis_srem("tags", "cache")

# Get cardinality
count = await redis_scard("tags")  # Returns 2

JSON Operations

from sillo_contrib.redis import redis_json_get, redis_json_set

# Set JSON data
user_data = {
    "name": "John Doe",
    "age": 30,
    "preferences": {"theme": "dark", "language": "en"}
}
await redis_json_set("user:123", ".", user_data)

# Get JSON data
user = await redis_json_get("user:123")
# Returns: {"name": "John Doe", "age": 30, "preferences": {...}}

# Get nested path
theme = await redis_json_get("user:123", ".preferences.theme")

Advanced Operations

from sillo_contrib.redis import redis_keys, redis_flushdb, redis_execute

# Get keys by pattern
user_keys = await redis_keys("user:*")

# Execute raw Redis commands
info = await redis_execute("INFO", "memory")

# Flush database (use with caution!)
await redis_flushdb()

Configuration

RedisConfig Options

from sillo_contrib.redis import RedisConfig

config = RedisConfig(
    url="redis://localhost:6379/1",           # Redis URL
    db=1,                                     # Database number
    password="your_password",                 # Authentication
    decode_responses=True,                    # Decode to strings
    encoding="utf-8",                         # String encoding
    socket_timeout=5.0,                       # Socket timeout
    socket_connect_timeout=5.0,               # Connection timeout
    socket_keepalive=True,                    # TCP keepalive
    health_check_interval=30,                 # Health check interval
    max_connections=20,                       # Connection pool size
    retry_on_timeout=False,                   # Retry on timeout
)

Environment Variables

Configure Redis using environment variables:

export REDIS_URL="redis://localhost:6379/1"
export REDIS_PASSWORD="your_secure_password"
export REDIS_DB="1"
export REDIS_DECODE_RESPONSES="true"
export REDIS_SOCKET_TIMEOUT="5.0"
export REDIS_MAX_CONNECTIONS="20"

Then load in code:

from sillo_contrib.redis import RedisConfig

config = RedisConfig.from_env()

Connection Management

Automatic Lifecycle

The Redis client automatically manages connections:

from sillo import silloApp
from sillo_contrib.redis import init_redis

app = silloApp()

# Initialize Redis - automatically handles startup/shutdown
init_redis(app, url="redis://localhost:6379")

# Redis client connects on first use and disconnects on app shutdown

Manual Connection Control

from sillo_contrib.redis import RedisClient, RedisConfig

config = RedisConfig(url="redis://localhost:6379")
client = RedisClient(config)

# Manual connection management
await client.connect()
try:
    await client.get("my_key")
finally:
    await client.close()

Health Monitoring

# Check connection health
from sillo_contrib.redis import get_redis

redis = get_redis()
is_healthy = await redis.ping()  # Returns True if connected

Error Handling

from sillo_contrib.redis import RedisConnectionError, RedisOperationError

try:
    value = await redis_get("my_key")
except RedisConnectionError as e:
    # Handle connection issues
    print(f"Connection failed: {e}")
except RedisOperationError as e:
    # Handle operation failures
    print(f"Operation failed: {e}")

Examples

Complete Application

from sillo import silloApp
from sillo.http import Request, Response
from sillo_contrib.redis import (
    init_redis, RedisDepend
)

app = silloApp()

# Initialize Redis
init_redis(app, url="redis://localhost:6379", db=1)

@app.get("/user/{user_id}")
async def get_user(
    request: Request,
    response: Response,
    user_id: str,
    redis: RedisClient = RedisDepend()
):
    # Try cache first
    cached = await redis.json_get(f"user:{user_id}")
    if cached:
        return {"user": cached, "source": "cache"}

    # Simulate database fetch
    user_data = {
        "id": user_id,
        "name": "John Doe",
        "email": "john@example.com"
    }

    # Cache for 5 minutes
    await redis.json_set(f"user:{user_id}", ".", user_data, ex=300)

    return {"user": user_data, "source": "database"}

@app.post("/user/{user_id}/visit")
async def record_visit(
    request: Request,
    response: Response,
    user_id: str,
    redis: RedisClient = RedisDepend()
):
    visit_count = await redis.incr(f"user:{user_id}:visits", 1)
    return {"user_id": user_id, "visits": visit_count}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Session Management

from sillo import silloApp
from sillo_contrib.redis import RedisDepend
import json
import uuid

app = silloApp()
init_redis(app)

@app.post("/login")
async def login(request: Request, response: Response, redis: RedisClient = RedisDepend()):
    user_data = await request.json

    # Create session
    session_id = str(uuid.uuid4())
    session_data = {
        "user_id": user_data["id"],
        "username": user_data["username"],
        "login_time": "2024-01-01T00:00:00Z"
    }

    # Store session in Redis (expire in 24 hours)
    await redis.json_set(f"session:{session_id}", ".", session_data, ex=86400)

    return {"session_id": session_id}

@app.get("/profile")
async def get_profile(
    request: Request,
    response: Response,
    redis: RedisClient = RedisDepend()
):
    # Get session ID from header/cookie (implement as needed)
    session_id = request.headers.get("X-Session-ID")

    if not session_id:
        return response.json({"error": "No session"}, status_code=401)

    # Get session data
    session_data = await redis.json_get(f"session:{session_id}")
    if not session_data:
        return response.json({"error": "Invalid session"}, status_code=401)

    return {"user": session_data}

Testing

Mock Redis Client

from unittest.mock import AsyncMock, patch
import pytest
from sillo_contrib.redis import get_redis

@pytest.fixture
def mock_redis():
    mock_client = AsyncMock()
    mock_client.get.return_value = "test_value"
    mock_client.set.return_value = True

    with patch('sillo_contrib.redis.get_redis', return_value=mock_client):
        yield mock_client

async def test_my_handler(mock_redis):
    # Your test code here
    pass

Test Utilities

# Test with real Redis (requires Redis server)
async def test_with_real_redis():
    from sillo_contrib.redis import init_redis, redis_set, redis_get

    app = silloApp()
    init_redis(app, url="redis://localhost:6379/15")  # Use test database

    await redis_set("test_key", "test_value")
    value = await redis_get("test_key")

    assert value == "test_value"

Performance Tips

1. Connection Pooling: Redis client uses connection pooling by default
2. Pipeline Operations: Use Redis pipelines for multiple operations
3. JSON Path Queries: Use specific JSON paths instead of root ”.” when possible
4. TTL Management: Set appropriate TTL values to prevent memory bloat
5. Key Naming: Use consistent key naming patterns for better organization

Troubleshooting

Common Issues

Connection Refused

# Check if Redis is running
redis-cli ping

# Verify connection details
redis-cli -h localhost -p 6379 ping

Authentication Failed

# Check password in Redis config
config = RedisConfig(
    url="redis://localhost:6379",
    password="your_correct_password"  # Ensure this matches Redis config
)

Memory Issues

# Monitor Redis memory usage
info = await redis_execute("INFO", "memory")

# Set appropriate TTL values
await redis_set("temp_key", "value", ex=300)  # 5 minutes

Debug Mode

Enable debug logging to troubleshoot issues:

import logging

logging.getLogger("sillo.redis").setLevel(logging.DEBUG)

Built with ❤️ by the @sillohq community.