Shulou(Shulou.com)05/31 Report--

Most people do not understand the knowledge points of this article "how to use the Schedule module of Python", so the editor summarizes the following content, detailed content, clear steps, and has a certain reference value. I hope you can get something after reading this article. Let's take a look at this "how to use the Schedule module of Python" article.

If you want to execute a Python script periodically on a Linux server, the most famous choice should be the Crontab script, but Crontab has the following disadvantages:

1. It is not convenient to perform a second-level task.

two。 When there are hundreds of scheduled tasks to be performed, the management of Crontab will be very inconvenient.

Another option is Celery, but the configuration of Celery is troublesome. If you just 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 as possible, easy to use, without external dependencies, and preferably able to accommodate all the basic functions of Crontab, then the Schedule module is your best choice.

Using it to schedule tasks may take only a few lines of code to feel:

Import scheduleimport timedef job (): print ("Isimm working...") schedule.every (10) .minutes.do (job) while True: schedule.run_pending () time.sleep (1)

The above code means that the job function is executed every 10 minutes, which is very simple and convenient. You only need to introduce the schedule module by calling scedule.every (time count). Time type .do (job) release cycle task.

The post-release periodic task needs to use the run_pending function to detect whether it is executed, so a While loop is required to constantly poll this function.

The following details about the installation and primary and advanced use of the Schedule module.

1. Prepare for

Please choose any of the following ways to enter the command to install dependencies:

1. Windows environment opens Cmd (start-run-CMD).

2. Open Terminal in MacOS environment (enter Terminal in command+ space).

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

Pip install schedule2. Basic use

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

Import scheduleimport timedef job (): print ("Ilemm working...") # execute task every ten minutes schedule.every (10) .minutes.do (job) # execute task schedule.every (). Hour.do (job) # execute task schedule.every (). Day.at ("10:30") .do (job) # execute task schedule.every (). Monday.do (job) every month # execute task at 13:15 every Wednesday schedule.every (). Wednesday.at ("13:15") .do (job) # execute task 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 a task only once, you can match it like this:

Import scheduleimport timedef job_that_executes_once (): # the task written here will only be executed once. Return schedule.CancelJobschedule.every (). Day.at ('22 job_that_executes_once 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 scheduledef greet (name): print ('Hello', name) # do () passes additional parameters to the job function schedule.every (2) .seconds.do (greet, name='Alice') schedule.every (4) .seconds.do (greet, name='Bob') to get all the current jobs

If you want to get all the current assignments:

Import scheduledef hello (): print ('Hello world') schedule.every () .second.do (hello) all_jobs = schedule.get_jobs () cancel all jobs

If some mechanism is triggered, you need to immediately clear all jobs from the current program:

Import scheduledef greet (name): print ('Hello {}' .format (name)) schedule.every () .second.do (greet) schedule.clear () tag function

When setting up a job, in order to manage the job later, you can tag the job so that you can get the job or cancel the job through tag filtering.

Import scheduledef greet (name): print ('Hello {}' .format (name)) # .tag is labeled 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 (tag): you can get all the tasks with this tag friends = schedule.get_jobs ('friend') # cancel all daily-tasks tagged tasks schedule.clear (' daily-tasks') set job deadlines

If you need to make an assignment due at a certain time, you can do it this way:

Import schedulefrom datetime import datetime, timedelta, timedef job (): print ('Boo') # run jobs every hour, stop schedule.every (1). Hours.until ("18:30") .do (job) # run jobs every hour, 2030-01-01 18:33 todayschedule.every (1). Hours.until ("2030-01-01 18:33") .do (job) # run jobs every hour Stop schedule.every (1) .hours.jobs (timedelta (hours=8)) .do (job) # run jobs every hour after 8 hours, stop schedule.every (1) .hours.jobs (time (11,33,42)) .do (job) # after 11:32:42, stop schedule.every (1) .hours.jobs (datetime (2020, 5, 17,11,36,20)) .do (job) after 2020-5-17 11:36:20

After the deadline, the job will not run.

Run all jobs immediately, regardless of their scheduling

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

Import scheduledef 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 with an interval of 10 seconds schedule.run_all (delay_seconds=10) 3. Advanced use of decorators to schedule jobs

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

From schedule import every, repeat, run_pendingimport time# this decorator effect is equivalent to schedule.every (10) .minutes.do (job) @ repeat (every (10) .decoration) def job (): print ("I am a scheduled job") while True: run_pending () time.sleep (1) parallel execution

By default, Schedule executes all jobs sequentially. The reason behind this is that it's hard to find a parallel execution model that makes everyone happy.

However, you can run each job in the form of multiple threads to address this limitation:

Import threadingimport timeimport scheduledef job1 (): print ("I'm running on thread% s"% threading.current_thread ()) def job2 (): print ("I'm running on thread% s"% threading.current_thread ()) def job3 (): print ("I'm running on thread% s"% threading.current_thread ()) def run_threaded (job_func): job_thread = threading.Thread (target=job_func) job_thread. Start () schedule.every (10) .seconds.do (run_threaded Job1) schedule.every (10) .seconds.do (run_threaded, job2) schedule.every (10) .seconds.do (run_threaded, job3) while True: schedule.run_pending () time.sleep (1) logging

The Schedule module also supports logging logging, which is used as follows:

Import scheduleimport logginglogging.basicConfig () schedule_logger = logging.getLogger ('schedule') # log level is DEBUGschedule_logger.setLevel (level=logging.DEBUG) def job (): print ("Hello, Logs") schedule.every () .second.do (job) schedule.run_all () schedule.clear ()

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 throws exceptions directly, which leads to a serious problem: all subsequent jobs are interrupted, so we need to catch these exceptions.

You can capture it manually, but some unexpected situations need to be captured automatically by the program, which can be done by adding a decorator:

Import functoolsdef catch_exceptions (cancel_on_failure=False): def catch_exceptions_decorator (job_func): @ functools.wraps (job_func) def wrapper (* args, * * kwargs): try: return job_func (* args * * kwargs) except: import traceback print (traceback.format_exc ()) if cancel_on_failure: return schedule.CancelJob return wrapper return catch_exceptions_decorator@catch_exceptions (cancel_on_failure=True) def bad_task (): return 1 / 0schedule.every (5) .minutes.do (bad_task)

In this way, any errors encountered by bad_task during execution will be caught by catch_exceptions, which is critical to ensuring the proper operation of the scheduling task.

The above is about the content of this article on "how to use the Schedule module of Python". I believe we all have a certain understanding. I hope the content shared by the editor will be helpful to you. If you want to know more about the relevant knowledge, please follow the industry information channel.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.