varco-redis

Redis backend for varco.

Provides two independent subsystems backed by Redis:

• RedisEventBus — Pub/Sub event bus implementing AbstractEventBus from varco_core
• RedisCache — async cache backend implementing CacheBackend from varco_core

Installation

pip install varco-redis
# or with uv:
uv add varco-redis

Event bus quick start

from varco_redis import RedisEventBus, RedisEventBusSettings
from varco_core.event import BusEventProducer, EventConsumer, listen, Event


# Define your events
class OrderPlacedEvent(Event):
    __event_type__ = "order.placed"
    order_id: str
    total: float


# Configure the bus
settings = RedisEventBusSettings(url="redis://localhost:6379/0")


async def main():
    async with RedisEventBus(settings) as bus:
        # --- Consumer side ---
        class OrderConsumer(EventConsumer):
            @listen(OrderPlacedEvent, channel="orders")
            async def on_placed(self, event: OrderPlacedEvent) -> None:
                print(f"Order placed: {event.order_id}")

        OrderConsumer().register_to(bus)

        # Give the Pub/Sub subscription time to establish
        import asyncio

        await asyncio.sleep(0.1)

        # --- Producer side ---
        producer = BusEventProducer(bus)
        await producer._produce(
            OrderPlacedEvent(order_id="abc", total=99.0),
            channel="orders",
        )

Event bus configuration

from varco_redis import RedisEventBusSettings

settings = RedisEventBusSettings(
    url="redis://redis.internal:6379/0",  # Redis connection URL
    channel_prefix="prod:",  # optional — "orders" → "prod:orders"
    socket_timeout=5.0,  # seconds, None = no timeout
)

Event bus lifecycle

# Explicit lifecycle
bus = RedisEventBus(settings)
await bus.start()  # connects to Redis, starts listener task
# ... use bus ...
await bus.stop()  # cancels listener, closes connection

# Context manager (recommended)
async with RedisEventBus(settings) as bus:
    ...

Wildcard subscriptions

Subscribing with channel=CHANNEL_ALL (the default) uses Redis PSUBSCRIBE "*" — the handler receives events from every channel published to this Redis instance. Use channel_prefix to scope channels to your service and avoid cross-service interference:

# All events with prefix "svc-a:" on this Redis
settings = RedisEventBusSettings(channel_prefix="svc-a:")
bus.subscribe(MyEvent, handler)  # receives from all "svc-a:*" channels

Cache quick start

from varco_redis.cache import RedisCache, RedisCacheSettings

cache_settings = RedisCacheSettings(
    url="redis://localhost:6379/0",
    key_prefix="myapp:",  # all keys stored as "myapp:<key>"
    default_ttl=300,  # seconds; None = no expiry
)

async with RedisCache(cache_settings) as cache:
    await cache.set("user:42", {"name": "Alice"})
    user = await cache.get("user:42")  # returns dict or None
    await cache.delete("user:42")
    await cache.clear()  # removes all "myapp:*" keys

Cache configuration

from varco_redis.cache import RedisCacheSettings

settings = RedisCacheSettings(
    url="redis://localhost:6379/0",
    key_prefix="prod:",
    default_ttl=600,
    socket_timeout=2.0,
)

Layered cache (L1 memory + L2 Redis)

from varco_core.cache import InMemoryCache, LayeredCache, TTLStrategy
from varco_redis.cache import RedisCache, RedisCacheSettings

l1 = InMemoryCache(strategy=TTLStrategy(60))  # fast in-process layer
l2 = RedisCache(RedisCacheSettings(key_prefix="app:"))  # shared Redis layer

async with LayeredCache(l1, l2, promote_ttl=60) as cache:
    await cache.set("product:1", product, ttl=300)
    # First read: L1 miss → L2 hit → promote to L1
    result = await cache.get("product:1")
    # Second read: served from L1 (no network round-trip)

Cache + CachedService with cross-process invalidation

from varco_core.cache import CachedService, LayeredCache, TTLStrategy, InMemoryCache
from varco_redis.cache import RedisCache, RedisCacheSettings
from varco_redis import RedisEventBus, RedisEventBusSettings

bus_settings = RedisEventBusSettings(url="redis://localhost:6379/0")
cache_settings = RedisCacheSettings(url="redis://localhost:6379/0", key_prefix="posts:")

async with (
    RedisEventBus(bus_settings) as bus,
    LayeredCache(
        InMemoryCache(strategy=TTLStrategy(60)),
        RedisCache(cache_settings),
        promote_ttl=60,
    ) as cache,
):
    cached = CachedService(
        post_service,
        cache,
        namespace="posts",
        default_ttl=300,
        bus=bus,  # publish invalidation events
        bus_channel="posts.invalidations",  # other nodes subscribe here
    )

    post = await cached.get(42)  # cached
    posts = await cached.list()  # cached
    await cached.update(42, {"title": "New"})  # invalidates + publishes event

Distributed rate limiting

RedisRateLimiter (varco_redis.rate_limit) implements varco_core.resilience.RateLimiter as a distributed sliding window over a Redis sorted set with an atomic Lua script — use it instead of InMemoryRateLimiter whenever more than one process/pod enforces the same limit (an in-memory limiter's counter is per-process, so N pods each allow the configured rate, not the total).

from varco_redis.rate_limit import RedisRateLimiter
from varco_core.resilience import RateLimitConfig

limiter = RedisRateLimiter(redis_client, RateLimitConfig(rate=100, period=1.0))


@rate_limit(limiter)
async def call_external() -> None: ...

Exported from varco_redis.__init__.

Distributed concurrency limiting

RedisBulkhead (varco_redis.bulkhead, Plan 005 Phase 8 / U-7's second leg) is the distributed sibling of varco_core.resilience.Bulkhead — a fleet-wide semaphore instead of a per-process one. Rate limiting and concurrency limiting are different primitives: a service can be well within its RedisRateLimiter budget and still overwhelm a downstream dependency with concurrent in-flight calls (e.g. a burst of long-running requests within the same second). Same public surface as Bulkhead (call()/protect()/available_slots()), backed by a Redis sorted set of holders scored by acquisition time and an atomic Lua acquire/release mirroring varco_redis.lock's token-guarded pattern. A holder that crashes without releasing has its slot reclaimed once slot_ttl elapses — no heartbeat process or cleanup job required.

from varco_redis.bulkhead import RedisBulkhead
from varco_core.resilience.bulkhead import BulkheadConfig

db_bulkhead = RedisBulkhead(BulkheadConfig(max_concurrent=10, max_wait=0.5))
await db_bulkhead.connect()

result = await db_bulkhead.call(fetch_user, user_id)


@db_bulkhead.protect
async def fetch_order(order_id: str) -> Order: ...

DI: RedisBulkheadConfiguration is an opt-in @Configuration (not auto-scanned by container.scan("varco_redis") — a bulkhead's max_concurrent is app-specific tuning, not a package-wide default):

from providify import DIContainer
from varco_redis.bulkhead import RedisBulkheadConfiguration

container = DIContainer()
await container.ainstall(RedisBulkheadConfiguration)
db_bulkhead = await container.aget(RedisBulkhead)

Exported from varco_redis.__init__.

DI integration

from providify import DIContainer
from varco_core.cache import CacheBackend
from varco_redis.cache import RedisCacheConfiguration

container = DIContainer()
await container.ainstall(RedisCacheConfiguration)

cache = await container.aget(CacheBackend)  # RedisCache singleton
await container.ashutdown()

Running tests

# Unit tests (no Redis required)
uv sync
uv run pytest

# Integration tests (requires Docker)
VARCO_RUN_INTEGRATION=1 uv run pytest -m integration

Connection settings

RedisConnectionSettings is a structured, env-var loadable config object that produces a URL and kwargs for redis.asyncio.

Plain connection

import redis.asyncio
from varco_redis.connection import RedisConnectionSettings

conn = RedisConnectionSettings(host="my-redis", port=6379, db=0)

client = redis.asyncio.from_url(conn.to_url(), **conn.to_redis_kwargs())
# to_url()  → "redis://my-redis:6379/0"

From environment variables

REDIS_HOST=my-redis
REDIS_PORT=6379
REDIS_DB=1

conn = RedisConnectionSettings.from_env()
client = redis.asyncio.from_url(conn.to_url(), **conn.to_redis_kwargs())

With password (AUTH)

conn = RedisConnectionSettings(host="my-redis", password="s3cret")
# to_url() → "redis://:s3cret@my-redis:6379/0"
client = redis.asyncio.from_url(conn.to_url(), **conn.to_redis_kwargs())

With ACL username + password (Redis 6+)

conn = RedisConnectionSettings(
    host="my-redis",
    username="alice",
    password="s3cret",
)
# to_url() → "redis://alice:s3cret@my-redis:6379/0"

With TLS / SSL

from varco_core.connection import SSLConfig
from pathlib import Path

ssl = SSLConfig(ca_cert=Path("/etc/ssl/redis-ca.pem"), verify=True)
conn = RedisConnectionSettings.with_ssl(ssl, host="prod-redis")
# to_url()           → "rediss://prod-redis:6379/0"
# to_redis_kwargs()  → {"decode_responses": False, "ssl": <SSLContext>}

client = redis.asyncio.from_url(conn.to_url(), **conn.to_redis_kwargs())

Or from env:

REDIS_HOST=prod-redis
REDIS_SSL__CA_CERT=/etc/ssl/redis-ca.pem
REDIS_SSL__VERIFY=true

With mTLS (client certificates)

ssl = SSLConfig(
    ca_cert=Path("/etc/ssl/ca.pem"),
    client_cert=Path("/etc/ssl/client.crt"),
    client_key=Path("/etc/ssl/client.key"),
)
conn = RedisConnectionSettings.with_ssl(ssl, host="prod-redis")

Connection settings reference

Note: RedisConnectionSettings is a general-purpose connection config. RedisEventBusSettings (used by RedisEventBus) is a separate, independent class — existing code that uses the event bus is unaffected.

Delivery semantics

Redis Pub/Sub provides at-most-once delivery — messages published while no subscriber is connected are silently dropped. If you need at-least-once or exactly-once delivery, use Redis Streams (planned as varco_redis.streams in a future release) or switch to varco_kafka.

Migration 1.x → 2.0

The no-op @Configuration aliases (RedisEventBusConfiguration, RedisChannelManagerConfiguration, RedisStreamConfiguration) were removed. Register the bus via scan instead:

# Before (1.x)
await container.ainstall(RedisEventBusConfiguration)

# After (2.0)
from varco_redis.di import bootstrap

bootstrap(container)  # or: container.scan("varco_redis", recursive=True)

The opt-in cache (RedisCacheConfiguration, RedisLayeredCacheConfiguration) is unchanged — still installed with await container.ainstall(...).