How to deal with recurrence dates in Django?

Using Django
1

I’m currently developing a Gym website and using django-recurrence to handle recurring training sessions.

However, I’m unsure how to work with recurrence dates in a Django QuerySet. Specifically, I want to sort trainings by the next upcoming date and filter trainings by a specific date, for example, to show sessions that occur on a selected day. What’s the best way to achieve this using django-recurrence?

Should I extract the next occurrence manually and store it in the model, or is there a more efficient approach for filtering and sorting recurring events? Any advice or examples would be greatly appreciated!

class Training(TimestampModel):
    template = models.ForeignKey(TrainingTemplate, on_delete=models.CASCADE, related_name='trainings')
    start_time = models.TimeField()
    end_time = models.TimeField()
    location = models.ForeignKey(TrainingLocation, on_delete=models.PROTECT, related_name='trainings')
    recurrences = RecurrenceField()
    members = models.ManyToManyField(Member, through=MemberTraining, related_name='trainings', blank=True)

    objects = TrainingManager()

I have a few ideas on how to handle this, but they all feel a bit over complicated.

  1. Add a next_training_date field to the Training model and use a Celery ETA task to update this field every time the training ends.

For example, when an admin creates a new training, the next_training_date is calculated on save. Then, a scheduled Celery task (with ETA) is created to update this date again after the current session ends.

While this might work, it feels like too much overhead just to support filtering and sorting by recurrence dates.

What would you recommend as a clean and efficient way to handle recurrence and QuerySet filtering using django-recurrence? Is there a better pattern or approach for dealing with recurring events in Django?

2

After the time I came up to the working solution. I decided to use Celery task to calculate next occurrence date, when training passed.

Future improvements can cover celery-beat schedule, to make sure all trainings was updated at the end of the day.

I can’t pretend that it is the best solution, but if anyone has more experience working with this type of work, I would appreciate any changes or improvements.

models.py

class Training(TimestampModel):
    template = models.ForeignKey(TrainingTemplate, on_delete=models.CASCADE, related_name='trainings')
    start_time = models.TimeField()
    end_time = models.TimeField()
    location = models.ForeignKey(TrainingLocation, on_delete=models.PROTECT, related_name='trainings')
    recurrences = RecurrenceField()
    next_occurrence = models.DateTimeField(null=True, editable=False)
    next_occurrence_celery_task_id = models.UUIDField(null=True, editable=False)
    members = models.ManyToManyField(Member, through=MemberTraining, related_name='trainings', blank=True)

    objects = TrainingManager()

    class Meta:
        ordering = ['next_occurrence', 'start_time']
        indexes = [
            models.Index(fields=['next_occurrence', 'start_time']),
        ]

    def save(self, *args, **kwargs):
        self.next_occurrence = self.calculate_next_occurrence()
        super().save(*args, **kwargs)

    def calculate_next_occurrence(self) -> datetime.datetime | None:
        try:
            now = timezone.now()
            today = timezone.make_naive(now).replace(hour=0, minute=0, second=0, microsecond=0)

            next_date = self.recurrences.after(today, inc=True)
            combined = datetime.datetime.combine(next_date.date(), self.start_time)
            aware_dt = timezone.make_aware(combined)

            if aware_dt > now:
                return aware_dt

            next_date = self.recurrences.after(today + timedelta(days=1), inc=True)
            return timezone.make_aware(datetime.datetime.combine(next_date.date(), self.start_time))

        except AttributeError:
            return None

signals.py

@receiver(post_save, sender=Training)
def calculate_next_occurrence(sender, instance, *args, **kwargs):
    if instance.next_occurrence:
        if instance.next_occurrence_celery_task_id:
            AsyncResult(str(instance.next_occurrence_celery_task_id)).revoke()

        result = calculate_next_training_occurrence.apply_async(
            args=(instance.id, ),
            eta=instance.next_occurrence,
            retry=False,
        )
        Training.objects.filter(id=instance.id).update(next_occurrence_celery_task_id=result.id)

tasks.py

@shared_task
def calculate_next_training_occurrence(training_id: int):
    try:
        training = Training.objects.get(id=training_id)
        training.next_occurrence = training.calculate_next_occurrence()
        training.save(update_fields=['next_occurrence'])

    except ObjectDoesNotExist:
        pass