Python celery beat tasks in external file
I am trying to setup python celery with beat scheduling of tasks. Only when the tasks are specified in the same file as where I setup celery it gets picked up by beat. This is my directory layout
├── celerybeat-schedule
├── __init__.py
└── project
├── celery.py
├── __init__.py
├── __pycache__
│ ├── celery.cpython-310.pyc
│ ├── __init__.cpython-310.pyc
│ └── tasks.cpython-310.pyc
└── tasks.py
celery.py
from celery import Celery
from celery.schedules import crontab
app = Celery(
"project",
backend="redis://localhost:6379/0",
broker="redis://localhost:6379/0",
include=["project.tasks"],
)
app.conf.update(
result_expires=3600,
)
#app.autodiscover_tasks(['project.certs'])
@app.on_after_finalize.connect
def setup_periodic_tasks(sender: Celery, **kwargs):
sender.add_periodic_task(10.0, test.s(1,2), name="1x2")
sender.add_periodic_task(10.0, project.tasks.test2.s(2,3), name="2**3")
@app.task
def test(x,y):
print(x*y)
return x*y
if __name__ == "__main__":
app.start()
tasks.py
from .celery import app
@app.task
def test2(x, y):
result = x ** y
print(result)
return result
when I start the worker it does know about the tasks from the other file
--- ***** -----
-------------- [queues]
.> celery exchange=celery(direct) key=celery
[tasks]
. project.celery.test
. project.tasks.test2
however beat doesn't schedule it when I start it. Any clues what I am missing