APScheduler

APScheduler#

APScheduler is a package that allows to start and controll processes that will do some jobs according to some triggers.For more check official APScheduler documentation.

from time import sleep
from unittest.mock import Mock

import apscheduler
from apscheduler.schedulers.background import BackgroundScheduler

Scheduler events#

There are a number of events that can happen with job. You can add some logic for when they occur by using listeners. Add listeners to the scheduler using the scheduler’s add_listener method.

The add_listener method typically takes the function to be executed and evnt code to triger execution.

The next cell implements the job code, the listener code, and creates the corresponding scheduler. Note that the listener was added with EVENT_JOB_EXECUTED, which declares some actions before a job is finished.

def my_listener(event):
    print("Job completed.")

def periodic_job():
    print("I'm a main task")

scheduler = BackgroundScheduler()
scheduler.add_job(periodic_job, "interval", seconds=3)
scheduler.add_listener(my_listener, apscheduler.events.EVENT_JOB_EXECUTED)

The following cell runs the scheduler we created.

scheduler.start()
sleep(10)
scheduler.shutdown()

I'm a main task
Job completed.
I'm a main task
Job completed.
I'm a main task
Job completed.

Each message produced by periodic_job is followed by message produced by the periodic_job proving that the job was terminated with the code specified in the listener.

Triggers#

Triggers determine when a job is executed.

Trigger properties are defined in the add_job method of the scheduler. There are two common ways to configure triggers for a job:

  • Pass an explicit object that describes the trigger as the trigger argument.

  • Pass a string that corresponds to the trigger type, which you can configure further.

The following cell shows a way to define triggers in the simplest way: by setting trigger="interval" and setting minutes parameter of the corresponding trigger.

scheduler = BackgroundScheduler()
scheduler.add_job(Mock(), trigger="interval", minutes=0.01)

<Job (id=a4a09732485342cda07deaf79f53279d name=Mock)>

The opposite way is to set a custom instance of IntervalTrigger to the trigger argument.

from apscheduler.triggers.interval import IntervalTrigger
scheduler.add_job(Mock(), trigger=IntervalTrigger(days=3))

<Job (id=266cd5c8d30348e693d4350369edad12 name=Mock)>

Each scheduler class has a _trigger_classes attribute that describes the correspondence of strings to actual classes to use. The following cell shows BackgroupdScheduler._trigger_classes.

BackgroundScheduler._trigger_classes

{'interval': apscheduler.triggers.interval.IntervalTrigger}
Contents