Вы можете выполнить шаги здесь для настройки celery для периодических задач.

В соответствии с этим вы можете сделать что-то приблизительно похожее на это:

import datetime
from celery import Celery
from myapp.models import Group


app = Celery()


@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # Setup and call send_reminders() every 60 seconds.

    sender.add_periodic_task(60.0, send_reminders, name='check reminders to be sent every minute')


@app.task
def send_reminders():
    # Celery task that gets all groups that needs reminders to be sent 30 minutes from now

    thirty_minutes_from_now = datetime.datetime.now() + datetime.timedelta(minutes=30)

    groups = Group.objects.filter(
        start_time__hour=thirty_minutes_from_now.hour, 
        start_time__minute=thirty_minutes_from_now.minute
    ).prefetch_related("joined")

    for group in groups:
        for member in group.joined.all():
            send_email_task.delay(member.email)


@app.task
def send_email_task(recipient):
    # Celery task to send emails

    send_mail(
        'group starting',
        'group starting',
        NOTIFICATION_EMAIL,
        [recipient],
        fail_silently=False
    )

Отказ от ответственности: Это не протестировано и не оптимизировано ;)

Почему периодические задачи не работают?

В Celery <= 5.2.7(stable-version) обнаружена ошибка.

Пусть периодические задания не работают.

Я исправил это в этом PR, вы можете отредактировать исходный код Celery как в этом PR, или попробовать Celery dev-версию.

Растворение 1

# your_app/task.py
@app.on_on_after_finalize.connect
def setup_periodic_tasks(sender, **kwargs):
    for group in Group.object.all():
        notification_time = group.start_time - timedelta(minutes=30)
        sender.add_periodic_task(clocked(notification_time),
                                 start_group_notification_task,
                                 kwargs={'recipients':group.recipients}
                                 name='send mail when group start time')
    # You need to connect the `Group.post_save` and `Group.post_delete` signal here,
    # to setup/revoke your periodic tasks when `Group` changed.

Вы можете использовать clocked в качестве вашего пользовательского класса планировщика, я процитировал его из django-celery-beat.

Солюшен 2

Возможно, вы можете использовать django-celery-beat, и создать m2m, связанные с вашими Group и PeriodicTask.

Выглядит проще.

Отказ от ответственности: Это тоже не протестировано и не оптимизировано.

Я смог понять, как запустить задачу на основе start_time, но все еще получаю некоторые ошибки, когда пытаюсь передать group в качестве аргумента для итерации через group.email_list.

Я добавил это в свой celery.py файл:

app.conf.beat_schedule = {
    'start_group_notification': {
        'task': 'start_group_notification_task',
        'schedule': crontab(),
    }
}

Задание запускается каждую минуту. Затем задача проверяет, есть ли у группы время начала в пределах 30 минут. В group/tasks.py

@shared_task(name='start_group_notification_task')
def start_group_notification_task():
    logger.info('sent email to whole group that group is starting')
    thirty_minutes_from_now = datetime.datetime.now() + datetime.timedelta(minutes=30)
    groups = Group.objects.filter(
        start_time__hour=thirty_minutes_from_now.hour, 
        start_time__minute=thirty_minutes_from_now.minute
    ).prefetch_related("joined")
    for group in groups: 
        send_mail (
                'group starting in 30 minutes',
                group.name,
                NOTIFICATION_EMAIL,
                ['me@gmail.com'],
                fail_silently=False
            )

Теперь это работает, но только для одного пользователя за раз. Идея заключается в том, чтобы задача выполнялась для каждого пользователя, который присоединился к группе. В ответе @Brian это было сделано, поэтому я реализовал это следующим образом:

@shared_task(name='start_group_notification_task')
def group_starting_in_30_task(group):
    logger.info('group starts in 30 email')
    for email in group.email_list:
        send_mail (
                'group starting in 30 minutes',
                group.name,
                NOTIFICATION_EMAIL,
                [email],
                fail_silently=False
            )

@shared_task(name='start_group_notification_task')
def start_group_notification_task():
    logger.info('sent email to whole group that group is starting')
    thirty_minutes_from_now = datetime.datetime.now() + datetime.timedelta(minutes=30)
    groups = Group.objects.filter(
        start_time__hour=thirty_minutes_from_now.hour, 
        start_time__minute=thirty_minutes_from_now.minute
    ).prefetch_related("joined")
    for group in groups: 
        group_starting_in_30_task(group)

Я думал, что это будет работать, но я продолжаю получать ошибки о позиционных аргументах. Я не вижу проблемы, потому что задача, вызываемая crontab(), не принимает никаких аргументов, она только вызывает другую задачу и передает ей group:

TypeError: start_group_notification_task() takes 0 positional arguments but 1 was given

Где давали одного, я думал, что он сам дает?

Я получаю эту ошибку в терминале celery (не в терминале celery-beat), но только когда группа находится в 30 минутах от запуска. Каждую вторую минуту задача выполняется, но None, потому что она фильтруется здесь:

    groups = Group.objects.filter(
        start_time__hour=thirty_minutes_from_now.hour, 
        start_time__minute=thirty_minutes_from_now.minute
    ).prefetch_related("joined")

Все части вроде бы работают, но перестают работать, когда я добавляю другую задачу и другой цикл for для отправки уведомления всем в группе email_list и я не понимаю, как это исправить на данный момент.