Taskiq Redis Integration

Warnings

• gotcha Failing to set `result_ex_time` or `result_px_time` in `RedisAsyncResultBackend` will cause task results to persist indefinitely in Redis, leading to unbounded memory growth and potential performance issues.
  Fix: Always configure either `result_ex_time` (seconds) or `result_px_time` (milliseconds) when initializing `RedisAsyncResultBackend`. For example: `RedisAsyncResultBackend(redis_url=REDIS_URL, result_ex_time=3600)`.
• deprecated The `RedisScheduleSource` is inefficient for high-volume or dynamic schedules as it performs a full `SCAN` of Redis keys, leading to slow performance. It has been deprecated.
  Fix: Migrate to `ListRedisScheduleSource` for scheduling. This source is designed for more efficient dynamic scheduling by storing schedules in lists, reducing the overhead of retrieving them.
• gotcha Using `PubSubBroker` (instead of `ListQueueBroker` or `RedisStreamBroker`) delivers messages to *all* subscribed workers, rather than distributing them to a single worker. It also does not support acknowledgements, meaning messages can be lost if a worker fails during processing.
  Fix: If you need messages to be processed exactly once by a single worker and require message durability, use `ListQueueBroker` (for a simple FIFO queue) or `RedisStreamBroker` (for more advanced stream processing with consumer groups and acknowledgements).
• breaking Version 1.2.0 of `taskiq-redis` updated its internal `redis` dependency, requiring `redis-py` version 7 or newer. This might cause compatibility issues if your project uses an older version of `redis-py`.
  Fix: Ensure your project's `redis` library dependency is updated to version 7 or higher. Check your `requirements.txt` or `pyproject.toml` and upgrade `redis` if necessary.
• gotcha When using `RedisStreamBroker`, instantiating it creates a new Redis connection pool. This can lead to inefficient resource usage if your application already has an existing `redis.asyncio.Redis` client and you wish to reuse its connection pool.
  Fix: While `taskiq-redis` does not directly expose a way to inject an existing connection pool into `RedisStreamBroker`'s constructor, you might consider manually interacting with Redis Streams (`XADD`) using your existing client if strict resource sharing is critical, or accept the broker's dedicated connection pool.
• gotcha Older versions of `RedisStreamBroker` (prior to 1.2.2) could experience infinite locking issues with the `xautoclaim` lock, potentially preventing tasks from being processed.
  Fix: Upgrade to `taskiq-redis` version 1.2.2 or newer, which includes a fix that adds a timeout to the `RedisStreamBroker`'s `xautoclaim` lock.
• gotcha Using `taskiq-redis` schedule sources (e.g., `RedisScheduleSource`, potentially `ListRedisScheduleSource` in certain configurations) with a Redis Cluster can lead to `NOGROUP` or `mget` errors due to tasks not being stored in the same key slot across cluster nodes.
  Fix: Careful consideration of Redis Cluster key distribution is needed. For scheduled tasks, manual hash tag assignment might be required, though this could lead to hot-spotting. For `RedisStreamBroker` specifically, ensure consumer groups are properly initialized across the cluster or consider alternative brokers for clustered environments if issues persist.

Install

• pip install taskiq taskiq-redis Install core library and Redis integration

Imports

• ListQueueBroker

  from taskiq_redis import ListQueueBroker

• RedisAsyncResultBackend

  from taskiq_redis import RedisAsyncResultBackend

• RedisStreamBroker

  from taskiq_redis import RedisStreamBroker

  The Stream broker is specifically named 'RedisStreamBroker'.
• ListRedisScheduleSource

  from taskiq_redis import ListRedisScheduleSource

  RedisScheduleSource is deprecated and inefficient for high-volume use; ListRedisScheduleSource is the recommended replacement.

Quickstart

This quickstart demonstrates how to set up a `taskiq-redis` broker and result backend, define a task, send it, and retrieve its result. It highlights the importance of setting an expiration time for results to manage Redis memory usage. It also includes an optional scheduler setup for periodic tasks.

import asyncio
import os
from taskiq import TaskiqScheduler
from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend, ListRedisScheduleSource

REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379')

# 1. Create a RedisAsyncResultBackend to store task results.
# IMPORTANT: Always set result_ex_time or result_px_time to prevent unbounded Redis memory usage.
redis_result_backend = RedisAsyncResultBackend(
    redis_url=REDIS_URL,
    result_ex_time=3600 # Results expire after 1 hour
)

# 2. Create a broker instance (e.g., ListQueueBroker for reliable single-consumer processing).
# Pass the result backend to the broker.
broker = ListQueueBroker(
    url=REDIS_URL,
    result_backend=redis_result_backend
)

# 3. Define a task using the broker's decorator.
@broker.task
async def my_simple_task(value: str) -> str:
    print(f"Executing task with value: {value}")
    await asyncio.sleep(1) # Simulate async work
    return f"Processed: {value.upper()}"

# 4. (Optional) Create a scheduler if you need scheduled tasks.
scheduler = TaskiqScheduler(
    broker=broker,
    sources=[ListRedisScheduleSource(url=REDIS_URL)]
)

async def main():
    # Startup the broker (and scheduler if used).
    await broker.startup()
    if 'TASKIQ_RUN_SCHEDULER' in os.environ:
        await scheduler.startup()

    # 5. Send a task to the broker.
    task = await my_simple_task.kiq("hello taskiq")
    print(f"Task sent with ID: {task.task_id}")

    # 6. Wait for the result.
    result = await task.wait_result(timeout=10)
    if result.is_err:
        print(f"Task failed: {result.error}")
    else:
        print(f"Task result: {result.return_value}")

    # Shutdown the broker (and scheduler).
    if 'TASKIQ_RUN_SCHEDULER' in os.environ:
        await scheduler.shutdown()
    await broker.shutdown()

if __name__ == "__main__":
    # To run this example:
    # 1. Ensure a Redis server is running (e.g., via Docker: docker run --name some-redis -p 6379:6379 -d redis)
    # 2. Save this code as 'my_app.py'.
    # 3. In one terminal, start the worker: taskiq worker my_app:broker
    # 4. In another terminal, run the script to send tasks: python my_app.py
    # 5. To enable scheduler, set TASKIQ_RUN_SCHEDULER=1 before running the script and worker:
    #    TASKIQ_RUN_SCHEDULER=1 python my_app.py
    #    TASKIQ_RUN_SCHEDULER=1 taskiq scheduler my_app:scheduler
    asyncio.run(main())