Django APScheduler

0.7.0 · active · verified Thu Apr 16

django-apscheduler is a Django application that provides a lightweight wrapper around APScheduler, enabling the scheduling and persistence of background jobs within a Django project using its ORM. It allows for managing scheduled tasks directly through the Django admin interface and is suitable for applications requiring basic scheduling features without external task queues like Celery. The current version is 0.7.0, and releases occur as needed to support newer Django/Python versions and address fixes.

Common errors

Warnings

Install

Imports

Quickstart

To set up `django-apscheduler`, first add `django_apscheduler` to your `INSTALLED_APPS`. Then, create a custom Django management command (e.g., `your_project_name/management/commands/runapscheduler.py`) to initialize and start the scheduler with the `DjangoJobStore`. This approach is recommended to ensure a single scheduler instance runs in a dedicated process, avoiding duplicate job executions in multi-process web server environments. Finally, run `python manage.py migrate` and then execute your custom command, typically managed by a process supervisor like systemd or Supervisor.

import logging
from django.conf import settings
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from django.core.management.base import BaseCommand
from django_apscheduler.jobstores import DjangoJobStore
from django_apscheduler.models import DjangoJobExecution
from django_apscheduler import util

logger = logging.getLogger(__name__)

def my_job():
    # Your job processing logic here...
    logger.info("My job is running!")

# The `close_old_connections` decorator ensures that database connections that have become
# unusable or are obsolete are closed before and after your job has run. You should use it
# to wrap any jobs that you schedule that access the Django database in any way.
@util.close_old_connections
def delete_old_job_executions(max_age=604_800):
    """This job deletes APScheduler job execution entries older than `max_age` from the database."""
    logger.info(
        "Deleting old job executions... (anything older than %s seconds)", max_age
    )
    DjangoJobExecution.objects.delete_old_job_executions(max_age)

class Command(BaseCommand):
    help = "Runs APScheduler."

    def handle(self, *args, **options):
        scheduler = BlockingScheduler(timezone=settings.TIME_ZONE)
        scheduler.add_jobstore(DjangoJobStore(), "default")

        scheduler.add_job(
            my_job,
            trigger=CronTrigger(second="*/10"),  # Every 10 seconds
            id="my_job",  # The `id` assigned to each job MUST be unique
            max_instances=1,
            replace_existing=True,
        )
        logger.info("Added job 'my_job'.")

        scheduler.add_job(
            delete_old_job_executions,
            trigger=CronTrigger(day_of_week="mon", hour="00", minute="00"),  # Midnight on Monday
            id="delete_old_job_executions",
            max_instances=1,
            replace_existing=True,
        )
        logger.info(
            "Added daily job: 'delete_old_job_executions'."
        )

        # Add a listener to log job executions and errors
        # register_events(scheduler)

        scheduler.start()
        logger.info("Scheduler started. Press Ctrl+C to exit.")

# To run this, save it as `your_project_name/management/commands/runapscheduler.py`
# Then, in your Django settings.py, add 'django_apscheduler' to INSTALLED_APPS.
# Run migrations: `python manage.py migrate`
# Start the scheduler: `python manage.py runapscheduler` (preferably in a dedicated process)

view raw JSON →