Django Tasks Scheduler

The worker process follows a specific lifecycle to ensure reliability and visibility:

1. Registration: Upon startup, the worker validates that the requested queue names exist in the system. It then creates a WorkerKey containing its ID, assigned queues, and status, and adds itself to the WorkerList with a current heartbeat.
2. Heartbeat: The worker continuously updates its last heartbeat to signal it is alive.
3. Job Polling: The worker loops through its assigned queueKeys using a zrange+zpop pattern to find the next available job name.
4. Execution Lifecycle:
   • When a job is found, the worker updates its status to busy.
   • It retrieves the job data and updates the job status to running.
   • After execution, it updates the job status to either done or failed.
   • Finally, it reverts its own status to idle.
5. Idle State: If no jobs are pending in any assigned queues, the worker enters a sleep state.

django-tasks-scheduler is a database-backed asynchronous task scheduler for Django. It uses a central Task model to manage scheduling and a broker-based system to handle execution.

Core Components:

• Scheduled Task: A Django model instance representing the intent to run code. Tasks are categorized by type: ONCE, REPEATABLE, or CRON.
• Job: A record in the broker containing the specific execution details (method reference, arguments, callbacks, and timeouts) required to run a task.
• Queue: A collection of registries within the broker that track the lifecycle of jobs: scheduled, queued, active, finished, failed, and canceled.
• Worker: A process that listens to queues for jobs to execute. It includes a thread for command listening and can run a Scheduler sub-process.
• Scheduler: A sub-process of the Worker that periodically checks the database for enabled tasks and moves them from the scheduled registry to the queued registry so workers can pick them up.

The system relies on two main loop-based processes to move tasks from the database to successful execution:

1. The Scheduler Loop

The Scheduler process runs a loop that:

1. Checks the database for enabled tasks that meet their scheduling criteria.
2. Creates a Job for those tasks and adds them to the scheduled registry in the broker.
3. Checks the scheduled registry and moves jobs into the queued registry to make them available for workers.
4. Sleeps for an interval defined by SCHEDULER_INTERVAL.

2. The Worker Loop

The Worker process runs a loop that:

1. Pulls the first available job from the queued registry.
2. Moves the job to the active registry.
3. Executes the job's method.
4. On Success: Executes success callbacks, moves the job to the finished registry, and writes the result.
5. On Failure: Executes failure callbacks, moves the job to the failed registry, and writes the result.

To integrate django-tasks-scheduler into your Django project, follow these four steps:

1. Add to INSTALLED_APPS: Include "scheduler" in your settings.py.
2. Configure SCHEDULER_CONFIG: Define a SchedulerConfiguration object in settings.py to set global execution parameters, broker types, and TTL (Time To Live) values.
3. Configure SCHEDULER_QUEUES: Define a dictionary mapping queue names to QueueConfiguration objects (e.g., specifying the Redis URL).
4. Register URLs: Include scheduler.urls in your project's urls.py to enable scheduler-related endpoints.
5. Run Migrations: Execute python manage.py migrate to set up the necessary database tables.

# settings.py
INSTALLED_APPS = [
    "scheduler",
]

from scheduler.types import SchedulerConfiguration, Broker, QueueConfiguration

SCHEDULER_CONFIG = SchedulerConfiguration(
    EXECUTIONS_IN_PAGE=20,
    SCHEDULER_INTERVAL=10,
    BROKER=Broker.REDIS,
    CALLBACK_TIMEOUT=60,
    DEFAULT_SUCCESS_TTL=600,
    DEFAULT_FAILURE_TTL=31536000,
    DEFAULT_JOB_TTL=600,
    DEFAULT_JOB_TIMEOUT=300,
    DEFAULT_WORKER_TTL=600,
    DEFAULT_MAINTENANCE_TASK_INTERVAL=600,
    DEFAULT_JOB_MONITORING_INTERVAL=30,
    SCHEDULER_FALLBACK_PERIOD_SECS=120,
)

SCHEDULER_QUEUES = {
    "default": QueueConfiguration(URL="redis://localhost:6379/0"),
}

# urls.py
from django.urls import path, include

urlpatterns = [
    path("scheduler/", include("scheduler.urls")),
]

You can manage and schedule tasks via the Django Admin interface under the Tasks Scheduler section.

Steps to schedule a task:

1. Navigate to Tasks and click Add.
2. Enter a unique Name.
3. Select a Task Type:
   • Repeatable task: Requires an Interval (and unit) and an optional Repeat count (empty means run forever).
   • Cron task: Requires a cron string and an optional Repeat count.
4. Enter the Callable using Python dot notation (e.g., myapp.jobs.count).
5. Select a Queue (defined in SCHEDULER_QUEUES in your settings).
6. Set the Scheduled time in UTC.

Optional Fields:

• at front: If selected, the job takes priority over existing queued jobs.
• Timeout: Maximum seconds allowed for execution (blank for infinite).
• Result TTL: Time to live for the job result (-1 for never expires, 0 for immediate deletion, n for $n$ seconds).

All scheduler settings are configured using two main dictionaries in your Django settings.py: SCHEDULER_CONFIG (for global scheduler behavior) and SCHEDULER_QUEUES (for defining specific queue connections).

Use SchedulerConfiguration to define global parameters like intervals, timeouts, and TTLs. Use QueueConfiguration to define how workers connect to specific queues (e.g., Redis hosts, ports, or URLs).

import os
from scheduler.types import SchedulerConfiguration, Broker, QueueConfiguration

SCHEDULER_CONFIG = SchedulerConfiguration(
    EXECUTIONS_IN_PAGE=20,
    SCHEDULER_INTERVAL=10,
    BROKER=Broker.REDIS,
    CALLBACK_TIMEOUT=60,
    DEFAULT_SUCCESS_TTL=10 * 60,
    DEFAULT_FAILURE_TTL=365 * 24 * 60 * 60,
    DEFAULT_JOB_TTL=10 * 60,
    DEFAULT_JOB_TIMEOUT=5 * 60,
)

SCHEDULER_QUEUES = {
    "default": QueueConfiguration(
        HOST="localhost",
        PORT=6379,
        DB=0,
        USERNAME="some-user",
        PASSWORD="some-password",
        CONNECTION_KWARGS={
            "ssl_cert_reqs": "required",
            "ssl": True,
        },
    ),
    "high": QueueConfiguration(URL=os.getenv("REDISTOGO_URL", "redis://localhost:6379/0")),
    "low": QueueConfiguration(HOST="localhost", PORT=6379, DB=0, ASYNC=False),
}