Django APScheduler

APScheduler for Django.

This little wrapper around APScheduler enables storing persistent jobs in the database using Django's ORM rather than requiring SQLAlchemy or some other bloatware.

Features in this project:

• Work on both python2.* and python3+
• Manage jobs from Django admin interface
• Monitor your job execution status: duration, exception, traceback, input parameters.

Installation

pip install django-apscheduler

Usage

• Add django_apscheduler to INSTALLED_APPS in your Django project settings:

  INSTALLED_APPS = (
    ...
    django_apscheduler,
  )

• Run migrations:

  ./manage.py migrate

• Instantiate a new scheduler as you would with APScheduler. For example:

  from apscheduler.schedulers.background import BackgroundScheduler
  
  scheduler = BackgroundScheduler()

• Instruct the scheduler to use DjangoJobStore:

  from django_apscheduler.jobstores import DjangoJobStore
  
  # If you want all scheduled jobs to use this store by default,
  # use the name 'default' instead of 'djangojobstore'.
  scheduler.add_jobstore(DjangoJobStore(), 'djangojobstore')

• If you want per-execution monitoring, call register_events on your scheduler:

    from django_apscheduler.jobstores import register_events
    register_events(scheduler)

  It provides the following interface:

• Register any jobs as you would normally. Note that if you haven't set DjangoJobStore as the 'default' job store, you'll need to include jobstore='djangojobstore' in your scheduler.add_job calls.

• Don't forget to give each job a unique id. For example:

  @scheduler.scheduled_job("interval", seconds=60, id="job")
  def job():
    ...

  or use custom decorator for job registration. It will give id automatically:

  from django_apscheduler.jobstores import register_job
  
  @register_job("interval", seconds=60)
  def job():
    ...

• Start the scheduler:

  scheduler.start()

A full example project can be found in the example dir. Code snippet:

import time

from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job

scheduler = BackgroundScheduler()
scheduler.add_jobstore(DjangoJobStore(), "default")

@register_job(scheduler, "interval", seconds=1)
def test_job():
    time.sleep(4)
    print("I'm a test job!")
    # raise ValueError("Olala!")

register_events(scheduler)

scheduler.start()
print("Scheduler started!")