Python сельдерей выполняет задачи во внешнем файле ⇐ Python

[Anonymous]

Я пытаюсь настроить Python celery с ритмичным планированием задач. Только когда задачи указаны в том же файле, где я настраиваю сельдерей, он подхватывается по такту. Это мой макет каталога

├── 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

когда я запускаю работника, он знает о задачах из другого файла
--- ***** -----
-------------- [queues]
.> celery exchange=celery(direct) key=celery

[tasks]
. project.celery.test
. project.tasks.test2

однако Beat не планирует это при запуске. Есть какие-нибудь подсказки, чего мне не хватает?
Кажется, это работает
app.conf.beat_schedule = {
'1*2': {
'task': 'project.celery.test',
'schedule': 10.0,
'args': (1, 2)
},
'2**3': {
'task': 'project.tasks.test2',
'schedule': 10.0,
'args': (2, 3)
},
}

а этого нет. Добавляется только тестовый, а не другой
@app.on_after_finalize.connect
def setup_periodic_tasks(sender: Celery, **kwargs):
sender.add_periodic_task(10.0, test.s(3,7), name="3*7_setup")
sender.add_periodic_task(10.0, project.tasks.test2.s(3,2), name="3**2_setup")


Подробнее здесь: https://stackoverflow.com/questions/798 ... ernal-file