Redis ListQueueBroker not triggering task in basic FastAPI setup

[ishaan-os]

Hi, I'm very interested in getting Taskiq setup for my FastAPI project but am having trouble setting it up for even a basic test. If anyone could take a look to tell me what i'm doing wrong that would be very helpful.

I'm trying to using Redis for both queuing tasks and as the results backend. It's working successfully as a results backend when I use the InMemoryBroker to kiq, but the task doesn't get triggered when I use ListQueueBroker to kiq. Not sure what I'm doing wrong.

Here's the relevant code in my main.py:

# doesnt work with this broker
broker = ListQueueBroker(settings.redis_url).with_result_backend(
    RedisAsyncResultBackend(settings.redis_url)
)

# works with this broker
# # broker = InMemoryBroker().with_result_backend(
#     RedisAsyncResultBackend(settings.redis_url)
# )

redis_client = Redis.from_url(settings.redis_url)

app = FastAPI()

taskiq_fastapi.init(broker, "app.main:app")

@broker.task()
async def add(x: int, y: int) -> int:
    logging.info("IN TASK") # logged when using in memory broker, not when using list queue broker
    await asyncio.sleep(5)
    return x + y

@app.on_event("startup")
async def startup_event():
    await redis.flushall() # clearing queue
    if not broker.is_worker_process:
        await broker.startup()

    task = await add.kiq(1, 2)
    current_keys = await redis_client.keys(*)
    logging.info(f"Current keys: {current_keys}") # empty when inmemorybroker, but message when list broker not picked up

    results = await task.wait_result() # completes when InMemoryBroker, gets hung when using list queue broker
    updated_keys = await redis_client.keys(*)
    print(f"Updated keys: {updated_keys}") # shows results message for completed task when using in memory broker

@app.on_event("shutdown")
async def shutdown_event():
    if broker.is_worker_process:
        await broker.shutdown()

Here's my docker-compose for reference as well:

services:
  dev:
    &app
    build:
      context: .
      target: development
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
    volumes:
      - ./app:/app
    ports:
      - 8000:8000
    environment:
      - APP_ENV=development
      - APP_SECRET=secret
      - APP_VERSION=dev123
      - REDIS_URL=redis://redis/
      - DB_URL=postgresql+asyncpg://postgres:postgres@db:5432/postgres
    env_file:
      - .env
    depends_on:
      - db
      - redis

  db:
    image: postgres:15.4
    expose:
      - 5432
    ports:
      - 5432:5432
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=postgres

  redis:
    image: bitnami/redis:6.2.5
    hostname: redis
    restart: always
    environment:
      ALLOW_EMPTY_PASSWORD: "yes"
    healthcheck:
      test: redis-cli ping
      interval: 1s
      timeout: 3s
      retries: 50

[kokoc30]

Your Compose file starts Uvicorn and Redis, but doesn't start a Taskiq worker. ListQueueBroker sends a message to Redis; it doesn't execute the task inside the API process. That explains the queued message remaining there and wait_result() waiting for a result that no worker has produced. InMemoryBroker executes in-process, so it hides that missing service.

Run a separate worker using the module path from your example:

taskiq worker app.main:broker

For Compose, add a worker service using the same image/build, mounted code, and REDIS_URL=redis://redis/ as the API, with that command. The imported module must register add on the same broker. The Taskiq usage guide describes this producer/worker split.

Also adjust the startup/shutdown handlers. The FastAPI integration runs app startup inside worker processes too, so don't submit and wait for your test task unconditionally there. A worker could end up waiting for a task before its consumer has started. Keep those handlers focused on the API-side broker lifecycle:

@app.on_event("startup")
async def startup_event():
    if not broker.is_worker_process:
        await broker.startup()


@app.on_event("shutdown")
async def shutdown_event():
    if not broker.is_worker_process:
        await broker.shutdown()

Notice the not in the shutdown condition; your example currently has the opposite condition. Remove the startup flushall() too, since restarting either process would otherwise clear queued work/results.

Once both processes are running, enqueue your test from a request handler or a separate producer, then await its result if needed. taskiq_fastapi.init(...) integrates dependencies/lifecycle; it does not launch the worker for you.

The integration documentation shows the lifecycle guards. Newer FastAPI applications should express the same lifecycle using a lifespan handler; the decorators above match the style in your existing example.