worker logs a Maybe you forgot to import it? when it receives a message from the queue

[devraj]

I have spotted this before and have managed to resolve and am back at seeing the same message being emitted by the worker. I am running the application (FastAPI based, all source code available here).

The worker is kicker off inside a container:

  # TaskIQ worker
  worker:
    container_name: ${PROJ_NAME}-worker
    build:
      context: .
      dockerfile: Dockerfile
    command: ["taskiq", "worker", "${PROJ_NAME}.broker:broker"]
    env_file:
      - .env.development
    restart: unless-stopped
    volumes:
      - ./src/${PROJ_NAME}:/opt/${PROJ_NAME}
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_healthy
      rabbitmq:
        condition: service_healthy

The broker is defined as per the examples and uses RabbitMQ and redis:

from .config import config

# See PyPI for more information about taskiq_redis
# https://pypi.org/project/taskiq-redis/
# 
# Redis is recommended as the results backend for the tasks
from taskiq_redis import RedisAsyncResultBackend

# RabbitMQ is recommended as the broker 
from taskiq_aio_pika import AioPikaBroker

redis_result_backend = RedisAsyncResultBackend(
    config.redis_dsn
)

broker = AioPikaBroker(
    config.amqp_dsn,
    result_backend=redis_result_backend
)

The FastAPI application in api.py definitely calls the startup and shutdown on the events:

# TaskIQ configurartion so we can share FastAPI dependencies in tasks
@app.on_event("startup")
async def app_startup():
    if not broker.is_worker_process:
        await broker.startup()

# On shutdown, we need to shutdown the broker
@app.on_event("shutdown")
async def app_shutdown():
    if not broker.is_worker_process:
        await broker.shutdown()

For the purposes of getting things working I have tasks defined here and a FastAPI endpoint kicks off the task.

Extract from tasks.py:

from sqlalchemy.ext.asyncio import AsyncSession
from taskiq import TaskiqDepends

from ...db import get_async_session
from ...broker import broker

@broker.task
async def send_account_verification_email(
    session: AsyncSession = TaskiqDepends(get_async_session)
) -> None:
    import logging
    logging.error("Kicking off send_account_verification_email")

Extract from create.py that kicks off the task.:

@router.get("/verify")
async def verify_user(request: Request):
    """Verify an account
    """
    await send_account_verification_email.kiq()
    return {"message": "hello world"}

When I kick off the events, the handler returns as expected, but the worker logs a message saying:

labs-worker         | [2023-04-26 00:22:30,824][taskiq.receiver.receiver][WARNING][worker-1] task "labs.routers.auth.tasks:send_account_verification_email" is not found. Maybe you forgot to import it?

I know I had this working at some point and obviously am missing something that is causing it to not work again. I thought I'd reach out here to see if anyone can point my error.

Appreciate you sparing your time.

[devraj]

I found it. I had to specifically reference the tasks package

command: ["taskiq", "worker", "${PROJ_NAME}.broker:broker", "${PROJ_NAME}.routers.auth.tasks"]

I have to investigate the -fsd cli option to ensure auto discovery of tasks.

Hopefully someone else finds this useful

[devraj]

The -fsd option for those interested.

    command: ["taskiq", "worker", "-fsd", "${PROJ_NAME}.broker:broker",]

the same also applies for the scheduler in case you come across that.

[s3rius]

Hi, @devraj, If you're going to use fastapi with taskiq, we recommend you using our integrations. If you use it, all of your tasks that you use in your application are going to be automatically imported. Moreover, you will be able to use all of your fastapi dependencies in tasks.

You can read more about it in documentation: https://taskiq-python.github.io/guide/taskiq-with-fastapi.html.

Thank you for using our project.

[devraj]

That ended up being it. At some point I had configured the middleware for FastAPI and then removed it thinking I won't need to share the dependencies and was wondering what made it stop working.

A good note for anyone using FastAPI.

I've removed the -fsd flag in our projects and and using the fastapi helpers.

Thanks for taking the time to answer my question.