redis-client-kit

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

[!TIP] Building this with an AI assistant? Hand it one page instead of the whole site: the public API of both the async and the sync side, what each extra turns on, every setting with the default it actually has, the rules that break code when they are broken, the mistakes models make with this API, and a map of which page to fetch for the rest. Every docs page is also served as raw Markdown at its own URL, and a Copy page button at the top of each one hands it straight to a chat window.

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}

With Dishka DI

from dishka import make_async_container
from redis_client_kit.providers import AsyncRedisProvider

container = make_async_container(
    AsyncRedisProvider(),
    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.