Django+ DRF + Celery: Schedule the same task for different objects in database in different time for each object

Question

I am working on an HR application, where a company has a branch and each branch has its working days policy in which the branch decides the hours of start and end of work day and the hour when the day is absence for employee if they didn't checked in and the weekend days,

class WorkingDaysPolicy(TimeStampedModel):

    WeekendDays = (
        (5, "Saturday"),
        (6, "Sunday"),
        (0, "Monday"),
        (1, "Tuesday"),
        (2, "Wednesday"),
        (3, "Thursday"),
        (4, "Friday"),
    )

    id = models.UUIDField(primary_key=True, editable=False, default=uuid.uuid4)
    branch = models.ForeignKey(
        Branch,
        on_delete=models.CASCADE,
        blank=True,
        null=True,
        related_name="branch_working_days",
    )
    number_of_daily_working_hrs = models.PositiveSmallIntegerField(
        _("Number of daily working hours")
    )
    weekend_days = MultiSelectField(
        _("Weekend days"), choices=WeekendDays, null=True, blank=True
    )
    normal_overtime_hourly_rate = models.FloatField(
        _("normal overtime hourly rate"), null=True, blank=True
    )
    day_starts_at = models.TimeField(_("Working day starts at"), blank=True, null=True)
    day_ends_at = models.TimeField(_("Working day ends at"), blank=True, null=True)
    absence_Starts_at = models.TimeField(_("Absence Starts at"), blank=True, null=True)

now I have a background task that must work exactly at the absence_Starts_at time and only on the working days.

I tried to import the WorkingDaysPolicy model on celery.py to loop over its objects and assign a task for each branch but the app crashed on starting the dev server and raised

raise AppRegistryNotReady("Apps aren't loaded yet.")

How can I run the same task at different times for each branch based on what is saved in the database.

Destaq · Accepted Answer · 2023-02-25 12:52:06Z

The error message "Apps aren't loaded yet" suggests that the code is attempting to use the "WorkingDaysPolicy" model before the Django application has fully loaded. This is likely due to the fact that the celery.py file is being imported before the Django application is fully initialized.

One solution to this problem is to delay the loading of the celery.py file until after the Django application is fully initialized. One way to achieve this is to use the ready() method in the AppConfig class of the Django application to load the celery application after the Django application is fully initialized.

Here is an example implementation that you can try:

Create a new file called apps.py in your Django application directory. Add the following code to the apps.py file:

from django.apps import AppConfig
    
    
class YourAppConfig(AppConfig):
    name = 'yourapp'
    
    def ready(self):
        # import the celery application here
        from yourapp.celery import app as celery_app

Update the init.py file in your Django application directory to import the YourAppConfig class:

default_app_config = 'yourapp.apps.YourAppConfig'

Move the code that imports the "WorkingDaysPolicy" model and schedule the task to a separate function in the celery.py file.
In the ready() method of the YourAppConfig class, import the function from the celery.py file and call it to schedule the task.

This should ensure that the celery.py file is only loaded after the Django application is fully initialized, and should resolve the "Apps aren't loaded yet" error.

Collectives™ on Stack Overflow

Django+ DRF + Celery: Schedule the same task for different objects in database in different time for each object

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related