wooden coyote
Hey! I am setting up some celery workers, and they need to access some logic from my main app
The structure looks like this
fastapi-server
/app/...
/celery_task
/tasks/...
/__init__.py
In the init file in the celery tasks folder i am trying to setup the PATH such that all the tasks in the tasks folder is able to use the app folder, this would make it so i wouldn't have to update all my tasks in the future
The following is the current init file setup
import sys
from pathlib import Path
# Ensure the project root (fastapi-server/) is on sys.path so that 'app' and
# all other local packages are importable inside Celery's ForkPoolWorker
# processes. This runs once when the celery_task package is first imported.
_project_root = str(Path(__file__).resolve().parent.parent)
if _project_root not in sys.path:
sys.path.insert(1, _project_root)
from celery import Celery
from celery.schedules import crontab
from core.config import config
celery_app = Celery(
"worker",
backend=config.CELERY_BACKEND_URL,
broker=config.CELERY_BROKER_URL,
)
celery_app.conf.task_routes = {
"celery_task.tasks.cleanup_old_errors.cleanup_old_errors": "maintenance-queue",
"celery_task.tasks.recur_overdue_tasks.recur_overdue_tasks": "maintenance-queue",
}
celery_app.conf.update(task_track_started=True)
# Add beat schedule
celery_app.conf.beat_schedule = {
'cleanup-old-errors-daily': {
'task': 'celery_task.tasks.cleanup_old_errors.cleanup_old_errors',
'schedule': crontab(hour="2", minute="0"),
},
'recur-overdue-tasks-hourly': {
'task': 'celery_task.tasks.recur_overdue_tasks.recur_overdue_tasks',
'schedule': crontab(minute="0"), # top of every hour
},
}
# Import tasks to register them
from celery_task.tasks import cleanup_old_errors, recur_overdue_tasks # noqa: F401, E402
The current solution i came up with is in regards to
limpid pollen
