Redis store¶

This page covers the store. For the Redis broker transport, see Redis transport.

RedisStore keeps task states, results and scheduler claims in Redis, so producers, workers and schedulers in different processes or on different machines share them.

Install¶

pip install 'asynk[redis]'
# or
uv add 'asynk[redis]'

Import it from its own module. Without the extra, this import raises an ImportError that tells you to install it:

from asynk.store.redis import RedisStore

The example needs a Redis server at redis://localhost:6379/0, so the documentation tests do not run it:

"""A store on Redis, shared by every process that uses the same prefix."""

import asyncio

from asynk import App, Worker
from asynk.store.redis import RedisStore
from asynk.transport.memory import memory_transport

producer, consumer = memory_transport()
app = App(
    "orders",
    producer=producer,
    consumer=consumer,
    store=RedisStore("redis://localhost:6379/0", prefix="asynk:orders:"),
)


@app.task()
async def add(a: int, b: int) -> int:
    return a + b


async def main() -> None:
    worker = Worker(app)
    run = asyncio.create_task(worker.run())

    handle = await add.delay(1, b=2)
    print("result:", await handle.result(timeout=5))

    worker.request_shutdown()
    await run


if __name__ == "__main__":
    asyncio.run(main())

Configuration¶

RedisStore(url=None, *, client=None, ttl=86400.0, prefix="asynk:", max_poll_interval=1.0)

• url or client: give exactly one. With url, the store builds the client and closes it on stop(). With client, a redis.asyncio.Redis you built, you own it and close it yourself.
• ttl: seconds a key lives. Every write resets it.
• prefix: text before every key. Give each app that shares a Redis its own, for example "asynk:billing:".
• max_poll_interval: the longest pause, in seconds, between the checks result() makes while it waits.

A client built from url uses a 5 second socket timeout and a 5 second connect timeout. Query parameters in the URL override them (redis://redis:6379/0?socket_timeout=10). For anything else, such as TLS settings or retries, pass your own client.

How it maps to Redis¶

The keys are asynk:state:<id>, asynk:result:<id> and asynk:claim:<task>:<time>, with your prefix in place of asynk:. Scheduler claims use SET NX, which Redis runs atomically, so several schedulers can share one store and exactly one publishes each slot.

Pitfalls¶

• wait polls. result() can lag by up to max_poll_interval after the worker records the result.
• start() does not contact Redis. A wrong URL or a Redis that is down shows up as logged store errors, not as a failed startup. The worker and scheduler carry on without the store, as described in Results and the store.
• A hung Redis slows every task and every delay(): each store call waits up to socket_timeout (5 seconds by default). A task pays that for its running state, its terminal state and its result write, and delay() pays it once for the pending state.
• The scheduler guarantee that exactly one scheduler publishes each slot holds only while Redis answers. When the claim fails, every scheduler publishes; see Scheduled tasks.
• TaskHandle.state() and result() raise the client's redis.exceptions.ConnectionError or TimeoutError when Redis is down. Readers see the error; workers and the scheduler carry on without the store.
• A client you pass in without socket timeouts lets a hung Redis stall a claim or a record.
• A client you pass in must use decode_responses=False: values are bytes. With True, get raises a TypeError.
• Redis Cluster clients (RedisCluster) are not supported, and wait does not use pub/sub wake-ups.
• A producer-only process runs no worker or scheduler, so call app.store.stop() when it is done to close the connection pool; see Store lifecycle.
• Records last only as long as Redis keeps them: its persistence settings and eviction policy apply.