Of Stack

Detailed explanation of python periodic task scheduling tool Schedule

  • 2021-12-13 08:56:28
  • OfStack
Directory 1. Prepare 2. Basic use of parameter passing to get all current jobs. Cancel all job labels. Set job deadline 3. Advanced use of decorator to arrange jobs to perform logging exception handling in parallel

If you want to periodically execute an Python script, the most famous option is the Crontab script, but Crontab has the following drawbacks:

1. It is inconvenient to perform second-level tasks.

2. When there are hundreds of scheduled tasks to be performed, the management of Crontab will be particularly inconvenient.

Another option is Celery, but the configuration of Celery is troublesome. If you only need a lightweight scheduling tool, Celery will not be a good choice.

If you want to use a lightweight task scheduling tool, and want it to be as simple and easy to use as possible, without external dependence, and it is best to accommodate all the basic functions of Crontab, then Schedule module is your best choice.

It may only take a few lines of code to use it to schedule tasks. Feel 1: 


import schedule
import time
def job():
    print("I'm working...")
schedule.every(10).minutes.do(job)
while True:
    schedule.run_pending()
    time.sleep(1)

The above code indicates that the job function is executed every 10 minutes, which is very simple and convenient. You only need to introduce the schedule module and issue periodic tasks by calling scedule. every (number of times). Time type. do (job).

Periodic tasks after publication require run_pending Function to detect whether it is executed, so 1 While This function is constantly polled in a loop.

Let's talk about the installation and primary and advanced use of Schedule module.

Step 1 Prepare

Before you start, to ensure that Python and pip have been successfully installed on your computer, please select one of the following ways to enter the command to install dependencies:

The Windows environment opens Cmd (Start-Run-CMD).

The MacOS environment opens Terminal (command + space input Terminal).

If you are using the VSCode editor or Pycharm, you can use Terminal directly at the bottom of the interface. 


pip install schedule

2. Basic use

The most basic usage has been mentioned at the beginning of this article. Here are more examples of scheduling tasks: 


import schedule
import time
def job():
    print("I'm working...")
#  Every 10 Perform tasks in minutes 
schedule.every(10).minutes.do(job)
#  Perform tasks every hour 
schedule.every().hour.do(job)
#  Daily 10:30 Perform a task 
schedule.every().day.at("10:30").do(job)
#  Perform tasks every month 
schedule.every().monday.do(job)
#  Every week 3 Adj. 13:15 Sub-execution tasks 
schedule.every().wednesday.at("13:15").do(job)
#  The first time per minute 17 Execute tasks in seconds 
schedule.every().minute.at(":17").do(job)
while True:
    schedule.run_pending()
    time.sleep(1)

As you can see, the above examples cover the configuration from month to second. But if you want to run only one task, you can match it like this 


import schedule
import time
def job_that_executes_once():
    #  Tasks written here will only execute 1 Times ...
    return schedule.CancelJob
schedule.every().day.at('22:30').do(job_that_executes_once)
while True:
    schedule.run_pending()
    time.sleep(1)

Parameter passing

If you have parameters to pass to the job to execute, you just need to do this: 


import schedule
def greet(name):
    print('Hello', name)
# do()  Pass additional parameters to the job Function 
schedule.every(2).seconds.do(greet, name='Alice')
schedule.every(4).seconds.do(greet, name='Bob')

Get all current jobs

If you want to get all the current assignments: 


import schedule
def hello():
    print('Hello world')
schedule.every().second.do(hello)
all_jobs = schedule.get_jobs()

Cancel all jobs

If some mechanism triggers, you need to immediately clear all jobs of the current program: 


import schedule
def greet(name):
    print('Hello {}'.format(name))
schedule.every().second.do(greet)
schedule.clear()

Label function

When setting up a job, for the convenience of managing the job later, you can label the job, so that you can get the job or cancel the job through tag filtering. 


import schedule
def greet(name):
    print('Hello {}'.format(name))
# .tag  Label 
schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend')
schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend')
schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')
schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest')
# get_jobs( Label ) You can get all the tasks for this label 
friends = schedule.get_jobs('friend')
#  Cancel all  daily-tasks  Tasks of labels 
schedule.clear('daily-tasks')

Setting Job Deadlines

If you need to make an assignment due at a certain time, you can use this method: 


import schedule
from datetime import datetime, timedelta, time
def job():
    print('Boo')
#  Run the job every hour, 18:30 Post-stop 
schedule.every(1).hours.until("18:30").do(job)
#  Run the job every hour, 2030-01-01 18:33 today
schedule.every(1).hours.until("2030-01-01 18:33").do(job)
#  Run the job every hour, 8 Stopped in hours 
schedule.every(1).hours.until(timedelta(hours=8)).do(job)
#  Run the job every hour, 11:32:42 Post-stop 
schedule.every(1).hours.until(time(11, 33, 42)).do(job)
#  Run the job every hour, 2020-5-17 11:36:20 Post-stop 
schedule.every(1).hours.until(datetime(2020, 5, 17, 11, 36, 20)).do(job)

After the deadline, the job will not run.

Run all jobs immediately, regardless of their schedule

If a mechanism triggers and you need to run all jobs immediately, you can call schedule.run_all() : 


import schedule
def job_1():
    print('Foo')
def job_2():
    print('Bar')
schedule.every().monday.at("12:40").do(job_1)
schedule.every().tuesday.at("16:40").do(job_2)
schedule.run_all()
#  Run all jobs immediately, every job interval 10 Seconds 
schedule.run_all(delay_seconds=10)

STEP 3 Advanced use

Decorator arrangement work

If you think this form of setting homework is too verbose, you can also use decorator mode: 


pip install schedule
0

Parallel execution

By default, Schedule executes all jobs in sequence. The reason behind this is that it is difficult to find a parallel execution model that makes everyone happy.

However, you can solve this limitation by running each job in a multi-threaded manner: 


pip install schedule
1

Logging

The Schedule module also supports logging logging using: 


pip install schedule
2

The effect is as follows: 


DEBUG:schedule:Running *all* 1 jobs with 0s delay in between
DEBUG:schedule:Running job Job(interval=1, unit=seconds, do=job, args=(), kwargs={})
Hello, Logs
DEBUG:schedule:Deleting *all* jobs

Exception handling

Schedule does not automatically catch exceptions, it will throw them directly when it encounters them, which will cause a serious problem: all subsequent jobs will be interrupted, so we need to catch these exceptions.

You can capture it manually, but some unexpected situations require the program to capture it automatically, and you can do it with a decorator: 


pip install schedule
4

In this way, bad_task Any errors encountered during execution will be catch_exceptions Capture, which is very critical to ensure the normal operation of scheduled tasks.

This is the end of our article. If you like today's Python practical tutorial, please keep an eye on it.

The above is the python cycle task scheduling tool Schedule use details, more about the cycle task scheduling tool Schedule information please pay attention to other related articles on this site!

Related articles: