APScheduler <https://github.com/agronholm/apscheduler>
_ for Django <https://github.com/django/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.
It is a fork of https://github.com/jarekwg/django-apscheduler project (thanks to author), which had some errors, no tests, no PyPI package.
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.
-
Add
django_apscheduler
toINSTALLED_APPS
in your Django project settings:INSTALLED_APPS = ( ... django_apscheduler, )
-
Run migrations:
./manage.py migrate
-
Instanciate 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)
-
Register any jobs as you would normally. Note that if you haven't set
DjangoJobStore
as the'default'
job store, you'll need to includejobstore='djangojobstore'
in yourscheduler.add_job
calls. -
Don't forget to give unique id for each job, for example:
@scheduler.schedule_job("interval", seconds=60, id="job") def job(): ...
-
Start the scheduler:
scheduler.start()
Full example project you can find in example dir. Code snippet:
import time
from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_events
scheduler = BackgroundScheduler()
scheduler.add_jobstore(DjangoJobStore(), "default")
@scheduler.scheduled_job("interval", seconds=10, id="test")
def test_job():
time.sleep(4)
print "I'm a test job!"
# raise ValueError("Olala!")
register_events(scheduler)
scheduler.start()
print "Scheduler started!"