[Python] Tasks not registered in Celery worker

I am using docker for my Fastapi apps and Traefik, i am trying to use Celery and Redis to do a simple data integration. i cant seem to get the tasks to get registered in the Celery worker because in the Celery worker log i got nothing listed under the [tasks] part, here is the log:

/usr/local/lib/python3.12/site-packages/celery/platforms.py:841: SecurityWarning: You're running the worker with superuser privileges: this is
absolutely not recommended!
Please specify a different user using the --uid option.
User information: uid=0 euid=0 gid=0 egid=0
warnings.warn(SecurityWarning(ROOT_DISCOURAGED.format(

-------------- celery@4e9d33f9893f v5.5.2 (immunity)
--- ***** -----
-- ******* ---- Linux-6.8.0-52-generic-x86_64-with-glibc2.36 2025-05-06 07:28:04
- *** --- * ---
- ** ---------- [config]
- ** ---------- .> app: worker:0x7e1a094f8ad0
- ** ---------- .> transport: redis://redis:6379/0
- ** ---------- .> results: redis://redis:6379/0
- *** --- * --- .> concurrency: 12 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
-------------- [queues]
.> celery exchange=celery(direct) key=celery

[tasks]
[2025-05-06 07:28:05,279: INFO/MainProcess] Connected to redis://redis:6379/0
[2025-05-06 07:28:05,283: INFO/MainProcess] mingle: searching for neighbors
[2025-05-06 07:28:06,296: INFO/MainProcess] mingle: all alone
[2025-05-06 07:28:06,319: INFO/MainProcess] celery@4e9d33f9893f ready.


It is confirmed when i try to hit the celeryupdateshiftschedule in the referensi.py and check the task id progress in get_task_status i got PENDING status. I had a hunch the issue might be either in the name or folder structure. I've been trying to change the tasks name from app.tasks.update_shift_schedule or tasks.update_shift_schedule or even import it in the celery worker but it still doesnt work. Please help me what else should i try to fix this?

This is my project structure:

└── app
└── routers
└── referensi.py
└── celery_worker.py
└── db.py
└── dependencies.py
└── helper.py
└── main.py
└── schema.py
└── tasks.py
└── .gitlab-ci.yml
└── docker-compose.yml
└── Dockerfile
└── README.md
└── requirements.txt


this is my celery_worker.py:

from celery import Celery
import tasks

celery_app = Celery(
"worker",
broker="redis://redis:6379/0",
backend="redis://redis:6379/0"
)

celery_app.conf.task_routes = {
"update_shift_schedule": {"queue": "default"},
}


This is my docker-compose.yml:

version: '3'
networks:
bmkgbr:
external: true
volumes:
nfs_backend:
external: true
services:
e-bmkg-master-service-dev:
build:
context: /home/sdm_backend/Apps/e-bmkg/e-bmkg-backend/e-bmkg-master-service-dev
image: e-bmkg-master-service-dev:v1.0.0
restart: always
labels:
- "traefik.http.routers.e-bmkg-master-service-dev.rule=Host(`e-bmkg-master-service-dev.docker.localhost`)"
- "traefik.http.services.e-bmkg-master-service-dev.loadbalancer.server.port=80"
networks:
- bmkgbr
environment:
- HRM_DATABASE_URL=${HRM_DATABASE_URL}
- CELERY_BROKER_URL=redis://redis:6379/0
volumes:
- nfs_backend:/document_library
#NEW
redis:
image: redis:7
ports:
- "6379:6379"
networks:
- bmkgbr
restart: always
celery_worker:
build:
context: .
container_name: celery-worker
command: celery -A app.celery_worker:celery_app worker --loglevel=info
networks:
- bmkgbr
volumes:
- .:/app
depends_on:
- redis
environment:
- CELERY_BROKER_URL=redis://redis:6379/0


This is my Dockerfile:

FROM python:3.12

RUN pip install fastapi[standard]==0.115.5
RUN pip install sqlmodel==0.0.22
RUN pip install pytz==2024.1
RUN pip install pydantic-settings==2.4.0
RUN pip install pymongo==4.8.0
RUN pip install python-dotenv==1.0.1
RUN pip install mysqlclient==2.2.4
RUN pip install pymysql==1.1.1
RUN pip install requests==2.32.3
RUN pip install httpx==0.28.1
RUN pip install python-keycloak==4.7.2
RUN pip install psycopg2==2.9.3
RUN pip install httpx==0.27.2
RUN pip install python-jose==3.3.0
#NEW
RUN pip install celery[redis]==5.5.2

RUN useradd --create-home gitlab-runner
USER gitlab-runner

COPY ./ /app
WORKDIR /app

CMD ["fastapi", "run", "app/main.py", "--port", "80", "--host", "0.0.0.0", "--proxy-headers"]
EXPOSE 80


This is the relevant part of my tasks.py code:

from sqlmodel import Session, text
from schema import master_shift_pegawai
from db import engine_hrm
import httpx
from helper import column_to_camel, format_permission
import datetime
import pytz
from datetime import datetime
from sqlalchemy.dialects.mysql import insert
from celery_worker import celery_app
import logging

@celery_app.task(name="update_shift_schedule",bind=True)
def update_shift_schedule(self):
logging.basicConfig(level=logging.INFO)
logging.info("Starting update_shift_schedule")

#and so on the rest of my code


This is the relevant part of my referensi.py code:

from tasks import update_shift_schedule
from celery.result import AsyncResult
from celery_worker import celery_app
from fastapi import APIRouter, HTTPException

router = APIRouter(
prefix="/v1/sdm",
tags=["sdm"]
)
@router.post("/celeryupdateshiftschedule")
async def post_celery_update_shift_schedule(api_key: str = Depends(get_api_key)):
task=update_shift_schedule.delay()
return {
"success": True,
"statusCode": 202,
"data": task.id,
"message": "Shift schedule update task has been started"
}

@router.get("/task-status/{task_id}")
async def get_task_status(task_id: str):
result = AsyncResult(task_id, app=celery_app)
response = {
"task_id": task_id,
"status": result.status,
}

if result.status == 'PROGRESS':
response['progress'] = result.info # contains meta like {'current': 3, 'total': 10}
elif result.status == 'SUCCESS':
response['result'] = result.result
elif result.failed():
response['error'] = str(result.result)

return response

Continue reading...