Redis

Use Redis from Flux tasks for rate limiting, signaling, and distributed locks.

Flux 0.56.0 has no native Redis integration. There is no Redis-backed cache (the cache in flux/cache.py is disk-backed dill), no Redis broker, no Redis event store. Redis is just a thing your tasks can call.

That said, Redis is a useful tool to have inside a workflow: for rate limiting an external API, coordinating between concurrent tasks, or holding a distributed lock during a critical section. This page covers the patterns that come up.

Install

pip install redis

redis-py 5.x is async-native. Older sync code still works inside async def, but the async client is the right default.

Connection pooling

Don’t open a connection per task call — let redis-py pool them:

import redis.asyncio as redis

_pool = redis.ConnectionPool.from_url(
    "redis://redis.internal:6379/0",
    max_connections=20,
    decode_responses=True,
)

def get_client() -> redis.Redis:
    return redis.Redis(connection_pool=_pool)

Define the pool at module scope. Each worker process gets one pool; concurrent tasks within the worker share it.

Rate limiting (token bucket)

A common need: cap calls to an external API at N per minute across all workers. Redis fits because every worker reads the same counter:

from flux import task

@task
async def rate_limit(key: str, limit: int, window_seconds: int) -> None:
    r = get_client()
    current = await r.incr(key)
    if current == 1:
        await r.expire(key, window_seconds)
    if current > limit:
        raise RuntimeError(f"Rate limit exceeded for {key}")

Call it before any expensive external request. The simple INCR + EXPIRE works for fixed windows; for sliding windows use a sorted-set with timestamps.

Cross-task signaling

When one task needs to wait on a side effect produced by another (often in a different execution), BLPOP is the path of least resistance:

@task
async def wait_for_signal(key: str, timeout_seconds: int = 60) -> str:
    r = get_client()
    result = await r.blpop(key, timeout=timeout_seconds)
    if result is None:
        raise TimeoutError(f"No signal on {key}")
    _, value = result
    return value

This blocks the task, not the worker — Flux can dispatch other tasks on the same worker while one is waiting on Redis. Set the timeout aggressively; an indefinite wait will hold a slot in the worker’s task pool.

Distributed locks

Use redis-py’s built-in lock helper rather than reinventing SET NX EX:

@task
async def with_lock(key: str, ttl_seconds: int = 30):
    r = get_client()
    async with r.lock(key, timeout=ttl_seconds, blocking_timeout=5):
        # critical section
        return await do_the_thing()

The lock has a TTL so a crashed task does not strand it. Pick timeout longer than the critical section can possibly take, and blocking_timeout short enough that a starved task fails fast.

What goes wrong

Lock leaks across retries. If a task acquires a lock and then the retry policy fires, the new attempt is a different coroutine and the lock context is gone. Acquire the lock inside the task, not around task().
decode_responses=False and bytes. Without decode_responses=True, everything Redis returns is bytes. Set it on the pool once, not per call.
Cluster mode needs RedisCluster. redis.Redis against a cluster works for most commands but fails on cross-slot operations. Use redis.cluster.RedisCluster if you’re on a real cluster.

Roadmap

A Redis-backed OutputStorage for medium-size task results (between “fits in the event log” and “warrants S3”) is a reasonable community contribution. The interface is three methods — see S3-compatible object storage for a worked example to copy.