Как заставить django-tenants-celery-beat планировать периодические задачи

Я немного новичок в Celery и Celery Beat, но у меня уже есть tenant_schemas_celery запущенный в моем проекте:

import os

from celery.schedules import crontab
from tenant_schemas_celery.app import CeleryApp as TenantAwareCeleryApp

from config import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')

app = TenantAwareCeleryApp()
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

В файле apps.periodic_testing.tasks.py у меня есть:

from celery.schedules import crontab
from celery.utils.log import get_task_logger
from django_tenants_celery_beat.utils import generate_beat_schedule

from config.celery import app

logger = get_task_logger(__name__)


@app.task
def add(x=1, y=2):
    z = x + y
    print(z)


app.conf.beat_schedule = generate_beat_schedule(
    {
        "tenant_task": {
            "task": "app.tasks.add",
            "schedule": crontab(minute=0-59),
            "tenancy_options": {
                "public": False,
                "all_tenants": True,
                "use_tenant_timezone": True,
            }
        },
        "hourly_tenant_task": {
            "task": "app.tasks.add",
            "schedule": crontab(minute=0-59),
            "tenancy_options": {
                "public": False,
                "all_tenants": True,
                "use_tenant_timezone": False,
            }
        },
    }
)

Но я не могу заставить планировщик работать. Есть идеи, что я могу делать неправильно?

Вернуться на верх