redis-client-kit

Production-ready Redis client with optional Pydantic settings, Prometheus metrics, and Dishka dependency injection support.

Features

Installation

Core only (zero dependencies except redis-py)

pip install redis-client-kit

With optional features

# Pydantic settings support
pip install redis-client-kit[settings]

# Prometheus metrics
pip install redis-client-kit[metrics]

# Dishka dependency injection
pip install redis-client-kit[providers]

# Everything
pip install redis-client-kit[all]

Requirements: Python 3.10+

Quick Start

Basic Usage (Core)

from redis_client_kit import create_async_redis_client
from redis_client_kit.config import RedisSettingsProtocol

# Define your settings (can be a dataclass, dict, or Pydantic model)
class MySettings:
    class connection:
        host = "localhost"
        port = 6379
        db = 0
        
        @staticmethod
        def get_password():
            return None
    
    class cluster:
        enabled = False
        nodes = None
        require_full_coverage = True
        read_from_replicas = False
    
    class pool:
        max_connections = 10
        socket_timeout = 5.0
        socket_connect_timeout = 5.0
        socket_keepalive = True
        socket_keepalive_options = None
    
    class retry:
        enabled = True
        max_attempts = 3
        backoff_base = 0.1
        backoff_cap = 1.0
    
    class ssl:
        enabled = False
        cert_reqs = None
        ca_certs = None
        certfile = None
        keyfile = None
    
    class response:
        decode_responses = True
        encoding = "utf-8"
    
    health_check_interval = 30

# Create client
settings = MySettings()
client = create_async_redis_client(settings)

# Use it
await client.set("key", "value")
value = await client.get("key")
print(value)  # "value"

# Clean up
await client.aclose()

With Pydantic Settings

from redis_client_kit import create_async_redis_client
from redis_client_kit.settings import BaseRedisSettings, RedisConnectionSettings

# Use BaseRedisSettings with custom connection settings
settings = BaseRedisSettings(
    key_prefix="myapp",
    connection=RedisConnectionSettings(
        host="localhost",
        port=6379,
    ),
)

client = create_async_redis_client(settings)
await client.set("key", "value")
await client.aclose()

With Metrics (Prometheus)

from redis_client_kit import create_async_redis_client
from redis_client_kit.protocols import RedisMetricsProtocol

class MyMetrics(RedisMetricsProtocol):
    def record_command(self, command: str, status: str, duration: float) -> None:
        # Record to Prometheus
        redis_command_duration.labels(command=command, status=status).observe(duration)
    
    def record_error(self, error_type: str) -> None:
        redis_errors_total.labels(error_type=error_type).inc()
    
    def record_pool_stats(self, pool_size: int, pool_checked_out: int) -> None:
        redis_pool_size.set(pool_size)
        redis_pool_checked_out.set(pool_checked_out)

metrics = MyMetrics()
client = create_async_redis_client(settings, metrics=metrics)

With Prometheus Metrics (Built-in)

from redis_client_kit import create_async_redis_client
from redis_client_kit.metrics import RedisMetrics

# Create metrics instance (with optional prefix)
metrics = RedisMetrics(prefix="myapp")

# Create instrumented client
client = create_async_redis_client(settings, metrics=metrics)

# Metrics are automatically collected:
# - myapp_redis_pool_size
# - myapp_redis_pool_checked_out
# - myapp_redis_commands_total{command, status}
# - myapp_redis_command_duration_seconds{command}
# - myapp_redis_connection_errors_total{error_type}

Prometheus registers a metric name once per process, so a second RedisMetrics(prefix="myapp") raises ValueError. get_redis_metrics(prefix="myapp") returns the one instance per prefix instead, which is what a test suite that rebuilds its container per test wants.

With Dishka DI

from dishka import make_async_container
from redis_client_kit.providers import AsyncRedisProvider, PrometheusRedisMetricsProvider

container = make_async_container(
    AsyncRedisProvider(provide_default_metrics=False),
    PrometheusRedisMetricsProvider(),  # RedisMetrics when settings.metrics_enabled, else None
    SettingsProvider(),  # Your settings provider
)

async with container() as ctx:
    redis_client = await ctx.get(AsyncRedisClient)
    await redis_client.set("key", "value")

Redis Cluster

from redis_client_kit.settings import BaseRedisSettings, RedisClusterSettings

settings = BaseRedisSettings(
    key_prefix="myapp",
    cluster=RedisClusterSettings(
        enabled=True,
        nodes=["node1:6379", "node2:6379", "node3:6379"],
    ),
)

client = create_async_redis_client(settings)

Synchronous Client

from redis_client_kit.sync import create_redis_client

# Same API, but synchronous
client = create_redis_client(settings)
client.set("key", "value")
value = client.get("key")
client.close()

Key Features

Zero Overhead by Default

When you don't provide metrics, you get plain redis-py clients with zero instrumentation overhead:

# No metrics = plain Redis client (no performance cost)
client = create_async_redis_client(settings, metrics=None)

# With metrics = instrumented client
client = create_async_redis_client(settings, metrics=my_metrics)

Health Checks

from redis_client_kit import check_async_redis_health

is_healthy = await check_async_redis_health(client)
if is_healthy:
    print("Redis is ready!")

# PING alone says healthy for a read-only replica and for a primary that is out of memory
# under noeviction. write_key adds a SET <key> 1 EX 60 after the ping, for a service that
# needs Redis for more than reads.
is_healthy = await check_async_redis_health(client, write_key="myapp:health")

Graceful Shutdown

from redis_client_kit import close_async_redis_client

# Safe cleanup with timeout and asyncio.shield
await close_async_redis_client(client)

SSL/TLS Support

from redis_client_kit.settings import RedisSSLSettings

settings = BaseRedisSettings(
    key_prefix="myapp",
    ssl=RedisSSLSettings(
        enabled=True,
        cert_reqs="required",
        ca_certs="/path/to/ca.pem",
        certfile="/path/to/cert.pem",
        keyfile="/path/to/key.pem",
    ),
)

Retry Logic

from redis_client_kit.settings import RedisRetrySettings

settings = BaseRedisSettings(
    key_prefix="myapp",
    retry=RedisRetrySettings(
        enabled=True,
        max_attempts=5,
        backoff_base=0.5,
        backoff_cap=2.0,
    ),
)

Documentation

Full documentation: bedrock-python.github.io/redis-client-kit

• For AI agents — the whole library on one page
• Quick Start Guide
• Configuration Guide
• Advanced Usage
• API Reference

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

License

Apache 2.0 — see LICENSE.