Schließen Sie neue Sellerie regelmäßige Aufgabe in django

Es ist keine Frage, sondern Hilfe, um diejenigen, die finden, dass die Erklärung von regelmäßigen Aufgaben in der Sellerie 4.0.1-Dokumentation ist schwer zu integrieren in django:
http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html#entries

kopieren-einfügen, Sellerie config-Datei main_app/celery.py:

from celery import Celery
from celery.schedules import crontab

app = Celery()

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # Calls test('hello') every 10 seconds.
    sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')

    # Calls test('world') every 30 seconds
    sender.add_periodic_task(30.0, test.s('world'), expires=10)

    # Executes every Monday morning at 7:30 a.m.
    sender.add_periodic_task(
        crontab(hour=7, minute=30, day_of_week=1),
        test.s('Happy Mondays!'),
    )

@app.task
def test(arg):
    print(arg)

Frage

Aber was ist, wenn wir nutzen django, und unsere Aufgaben sind in einer anderen app? Mit Sellerie 4.0.1 haben wir nicht mehr @periodic_task Dekorateur. Also lasst uns sehen, was wir tun können.

Ersten Fall

Wenn Sie es vorziehen, halten Aufgaben und Ihren Zeitplan nah beieinander:

main_app/some_app/tasks.py

from main_app.celery import app as celery_app

@celery_app.on_after_configure.connect
    def setup_periodic_tasks(sender, **kwargs):
        # Calls test('hello') every 10 seconds.
        sender.add_periodic_task(10.0, test.s('hello'))

@celery_app.task
def test(arg):
    print(arg)

Können wir beat im debug-Modus:

celery -A main_app beat -l debug

und wir werden sehen, dass es gibt keine solche regelmäßige Aufgabe.

Zweiten Fall

Können wir versuchen zu beschreiben, was regelmäßige Aufgaben in der config-Datei wie folgt:

main_app/celery.py

...
app = Celery()

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # Calls test('hello') every 10 seconds.
    from main_app.some_app.tasks import test
    sender.add_periodic_task(10.0, test.s('hello'))
...

Das Ergebnis ist das gleiche. Aber es wird sich anders Verhalten, dass Sie sehen können, mit manuellem debugging über pdb. Im ersten Beispiel setup_periodic_tasks callback nicht ausgelöst wird, auf alle. Aber im zweiten Beispiel erhalten wir django.core.exceptions.AppRegistryNotReady: Apps aren't loaded yet. (diese Ausnahme wird nicht ausgedruckt)