Python uses the apscheduler module to set up the implementation of scheduled tasks

This article mainly introduces the implementation of Python using the apscheduler module to set scheduled tasks. The article introduces it in detail through sample code and has certain reference value. Interested friends can refer to it.

1. Installation

pip install apscheduler

2. Introduction to ApScheduler

1 Components of APScheduler

triggers: triggers
Triggers contain the scheduling logic of task execution, which determines the logic according to which tasks are executed regularly.

job stores; task storage
Scheduled tasks stored

executors: executors
Use cases perform tasks, including the creation and invocation of thread pools and process pools, etc.

schedulers: scheduler
It belongs to the control plane and plays the role of organizing other aspects.

2 Types of schedulers

There are several common types of schedulers. Among them, the most commonly used BackgroundScheduler is the non-blocking type, because under normal circumstances, scheduled tasks will be placed in web services. If the blocking type is used, the web service cannot be started, and the non-blocking type is used. Blocking type, after setting the scheduled task, ignore it and continue to execute the subsequent web service. As long as the web service is running, the scheduled task will always be valid.

  • BlockingScheduler: blocking

  • BackgroundScheduler: non-blocking (running in the background)

  • AsyncIOScheduler: Used when using the asyncio module

  • GeventScheduler: Used when using the gevent module

  • TornadoScheduler: used when building Tornado applications

  • TwistedScheduler: used when building Twisted applications

  • QtScheduler: used when building Qt applications

3 Built-in trigger types

  • date: used when executing once at a certain point in time

  • interval: used when executing a fixed time interval loop

  • cron: used when executing at a specific time of day

  • calendarinterval: Used when you want to execute at a specific time of day or within a calendar-based time interval

3. Usage examples

Here we take the non-blocking BackgroundScheduler scheduler as an example.

1 Use date type triggers

As follows, three methods of setting date and time are used

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time())) + " in do func : Name: " + name + "Age:" + str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # Set the specified date for execution by date
    sched.add_job(do_func,trigger="date",run_date=date(2022,5,25),args=("Zhang Sanfeng",100))

    # Through datetime, set the specified date and execute at the specified time
    sched.add_job(do_func, trigger="date", run_date=datetime(2022, 5, 25,14,0,10), args=("Zhang Sanfeng", 100))

    # Specify date and timetable directly using text
    sched.add_job(do_func, trigger="date", run_date="2022-05-25 14:0:20", args=("Zhang Sanfeng", 100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

From the execution results, it can be found that the first date specified by date is executed at 0 o’clock by default. Obviously the time has passed and will not be executed. The second and third ones are executed at the specified time point. Something else to note here. Yes, it can be seen from printing that after the execution of the main function is completed, the while loop print statement under the main function has been executed, and during the execution of the loop, the scheduled task is still in effect. This is the principle of the non-blocking scheduler. If it is blocking, in this code, it will always be stuck in the main function. The while loop statement under main will not be executed. Therefore, in actual use, there are many non-blocking applications.

2 Use interval type triggers

The following code demonstrates the use of time interval loop execution.

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time())) + " in do func : Name: " + name + "Age:" + str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # Execute every 3 seconds
    sched.add_job(do_func,trigger="interval",args=("Zhang Sanfeng",100),seconds=3)
    # Execute every 3 minutes
    sched.add_job(do_func, trigger="interval", args=("Zhang Sanfeng", 100), minutes=3)
    # Execute every 3 hours
    sched.add_job(do_func, trigger="interval", args=("Zhang Sanfeng", 100), hours=3)
    # Execute every 3 days
    sched.add_job(do_func, trigger="interval", args=("Zhang Sanfeng", 100), days=3)
    # Execute every 3 weeks
    sched.add_job(do_func, trigger="interval", args=("Zhang Sanfeng", 100), weeks=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

Because the time span in the above code is relatively large, here we only demonstrate the code that is executed once every 3 seconds.

code show as below:

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time())) + " in do func : Name: " + name + "Age:" + str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # Execute every 3 seconds
    sched.add_job(do_func,trigger="interval",args=("Zhang Sanfeng",100),seconds=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

3 Use cron type triggers

The cron trigger is somewhat similar to the use of the crontab timer on Linux. The code is as follows:

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time())) + " in do func : Name: " + name + "Age:" + str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # The task will be triggered on the third Friday of June, July, August, November and December at 00:00, 01:00, 02:00 and 03:00
    sched.add_job(do_func,trigger="cron",month='6-8,11-12', day='3rd fri', hour='0-3',args=( "Zhang Sanfeng",100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

It should be noted here that unnecessary fields can be omitted. When the time parameter is omitted, the parameter before the explicitly specified parameter will be set to *, and the subsequent parameters will be set to the minimum value. The minimum value of week and day_of_week is *

day=1, minute=20
Equivalent to
year='*', month='*', day=1, week='*', day_of_week='*', hour='*', minute=20, second= 0

In addition, you can also use crontab expressions directly, as follows:

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time())) + " in do func : Name: " + name + "Age:" + str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # The task will be triggered on the third Friday of June, July, August, November and December at 00:00, 01:00, 02:00 and 03:00
    sched.add_job(do_func,trigger=CronTrigger.from_crontab('48 10 1-15 sep-nov *'),args=("Zhang Sanfeng",100))

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

4. How to use decorators for timers

Taking the code executed in a time interval loop as an example, the following is the method without using decorators.

from apscheduler.schedulers.background import BackgroundScheduler
import time
from datetime import date
from datetime import datetime


def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time())) + " in do func : Name: " + name + "Age:" + str(age))

def main():
    print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    sched=BackgroundScheduler()

    # Execute every 3 seconds
    sched.add_job(do_func,trigger="interval",args=("Zhang Sanfeng",100),seconds=3)

    sched.start()

if __name__=="__main__":
    main()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

Modify to use a decorator as follows:

from apscheduler.schedulers.background import BackgroundScheduler
import time

sched=BackgroundScheduler()

@sched.scheduled_job(trigger="interval",args=("Zhang Sanfeng",100),seconds=3)
def do_func(name,age):
    print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(time.time())) + " in do func : Name: " + name + "Age:" + str(age))

if __name__=="__main__":
    sched.start()
    while True:
        print(time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
        time.sleep(1)

This concludes this article on the implementation of scheduled tasks in Python using the apscheduler module.

Take action, it is better to be on the road than to wait and see all the time. In the future, you will definitely thank yourself for working hard now! If you want to learn and improve but can’t find the information and there is no one to answer your questions, please join the group in time: 786229024. There are various test and development materials and technologies in which you can communicate together.

Finally: The complete software testing video tutorial below has been compiled and uploaded. Friends who need it can get it by themselves[Guaranteed 100% Free]

Software testing interview document

We must study to find a high-paying job. The following interview questions are the latest interview materials from first-tier Internet companies such as Alibaba, Tencent, Byte, etc., and some Byte bosses have given authoritative answers. After finishing this set I believe everyone can find a satisfactory job based on the interview information.

The knowledge points of the article match the official knowledge files, and you can further learn relevant knowledge. Python entry skill treeHomepageOverview 388,289 people are learning the system