Sangat praktikal! Modul jadual, artifak tugas berkala Python!

Jika anda ingin melaksanakan skrip Python secara berkala pada pelayan Linux, pilihan yang paling terkenal hendaklah skrip Crontab, tetapi Crontab mempunyai kelemahan berikut:

​<strong>​1.不方便执行<strong>秒级的任务</strong>。​</strong>​

​<strong>​2.当需要执行的定时任务有上百个的时候，Crontab的<strong>管理就会特别不方便</strong>。​</strong>​

Pilihan lain ialah Saderi , tetapi konfigurasi Saderi lebih menyusahkan Jika anda hanya memerlukan alat penjadualan yang ringan, Saderi tidak akan menjadi pilihan yang baik.

Jika anda ingin menggunakan alat penjadualan tugas yang ringan, dan berharap ia semudah dan semudah mungkin untuk digunakan tanpa kebergantungan luaran, dan sebaik-baiknya boleh menampung semua fungsi asas Crontab, maka modul Jadual adalah untuk anda Pilihan yang jelas.

Menggunakannya untuk menjadualkan tugasan mungkin hanya memerlukan beberapa baris kod, rasakan:

# Python 实用宝典
import schedule
import time

def job():
    print("I'm working...")

schedule.every(10).minutes.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

Kod di atas bermaksud fungsi kerja dilaksanakan setiap 10 minit, iaitu sangat mudah dan selesa. Anda hanya perlu memperkenalkan modul jadual dan menerbitkan tugasan berkala dengan memanggil ​<strong>​scedule.every(时间数).时间类型.do(job)​</strong>​ .

Tugas berkala selepas keluaran perlu menggunakan fungsi ​<strong>​run_pending​</strong>​ untuk mengesan sama ada ia dilaksanakan, jadi <🎜 diperlukan > ​​<strong>​While​</strong> Gelung terus meninjau fungsi ini.

Yang berikut akan membincangkan pemasangan dan penggunaan asas dan lanjutan modul Jadual.

1 Persediaan

Sila pilih mana-mana kaedah berikut untuk memasukkan arahan untuk memasang dependensi :

1. Buka Cmd (Start-Run-CMD) dalam persekitaran Windows.

2. Persekitaran MacOS Open Terminal (perintah+ruang untuk memasuki Terminal).

3 Jika anda menggunakan editor VSCode atau Pycharm, anda boleh terus menggunakan Terminal di bahagian bawah antara muka.

pip install schedule

2. Penggunaan asas

Penggunaan paling asas telah disebutkan pada permulaan artikel Berikut adalah lebih banyak contoh tugas penjadualan:

# Python 实用宝典
import schedule
import time

def job():
    print("I'm working...")

# 每十分钟执行任务
schedule.every(10).minutes.do(job)
# 每个小时执行任务
schedule.every().hour.do(job)
# 每天的10:30执行任务
schedule.every().day.at("10:30").do(job)
# 每个月执行任务
schedule.every().monday.do(job)
# 每个星期三的13:15分执行任务
schedule.every().wednesday.at("13:15").do(job)
# 每分钟的第17秒执行任务
schedule.every().minute.at(":17").do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

Seperti yang anda lihat, konfigurasi dari bulan ke saat diliputi oleh contoh di atas. Tetapi

Jika anda mahu menjalankan tugas sekali sahaja, anda boleh mengkonfigurasinya seperti ini:

# Python 实用宝典
import schedule
import time

def job_that_executes_once():
    # 此处编写的任务只会执行一次...
    return schedule.CancelJob

schedule.every().day.at('22:30').do(job_that_executes_once)

while True:
    schedule.run_pending()
    time.sleep(1)

Melalui parameter

Jika anda mempunyai parameter yang perlu dihantar kepada Untuk melaksanakan kerja, anda hanya perlu melakukan ini:

# Python 实用宝典
import schedule

def greet(name):
    print('Hello', name)

# do() 将额外的参数传递给job函数
schedule.every(2).seconds.do(greet, name='Alice')
schedule.every(4).seconds.do(greet, name='Bob')

Dapatkan semua kerja semasa

Jika anda ingin mendapatkan semua kerja semasa:

# Python 实用宝典
import schedule

def hello():
    print('Hello world')

schedule.every().second.do(hello)

all_jobs = schedule.get_jobs()

Batalkan semua kerja

Jika beberapa mekanisme dicetuskan, anda perlu mengosongkan semua kerja program semasa dengan segera:

# Python 实用宝典
import schedule

def greet(name):
    print('Hello {}'.format(name))

schedule.every().second.do(greet)

schedule.clear()

Fungsi teg

Apabila menyediakan kerja, untuk memudahkan pengurusan tugasan berikutnya, anda boleh menandai tugas, supaya anda boleh menapis kerja dengan menapis atau membatalkan kerja itu.

# Python 实用宝典
import schedule

def greet(name):
    print('Hello {}'.format(name))

# .tag 打标签
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(标签)：可以获取所有该标签的任务
friends = schedule.get_jobs('friend')

# 取消所有 daily-tasks 标签的任务
schedule.clear('daily-tasks')

Tetapkan tarikh akhir tugasan

Jika anda perlu membuat tugasan yang perlu dibayar pada masa tertentu, anda boleh menggunakan kaedah ini:

# Python 实用宝典
import schedule
from datetime import datetime, timedelta, time

def job():
    print('Boo')

# 每个小时运行作业，18:30后停止
schedule.every(1).hours.until("18:30").do(job)

# 每个小时运行作业，2030-01-01 18:33 today
schedule.every(1).hours.until("2030-01-01 18:33").do(job)

# 每个小时运行作业，8个小时后停止
schedule.every(1).hours.until(timedelta(hours=8)).do(job)

# 每个小时运行作业，11:32:42后停止
schedule.every(1).hours.until(time(11, 33, 42)).do(job)

# 每个小时运行作业，2020-5-17 11:36:20后停止
schedule.every(1).hours.until(datetime(2020, 5, 17, 11, 36, 20)).do(job)

Selepas tarikh akhir, kerja tidak akan berjalan.

Jalankan semua kerja dengan segera tanpa mengira jadual mereka

如果某个机制触发了，你需要立即运行所有作业，可以调用 ​<strong>​schedule.run_all()​</strong>​ :

# Python 实用宝典
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()

# 立即运行所有作业，每次作业间隔10秒
schedule.run_all(delay_seconds=10)

3.高级使用

装饰器安排作业

如果你觉得设定作业这种形式太啰嗦了，也可以使用装饰器模式：

# Python 实用宝典
from schedule import every, repeat, run_pending
import time

# 此装饰器效果等同于 schedule.every(10).minutes.do(job)
@repeat(every(10).minutes)
def job():
    print("I am a scheduled job")

while True:
    run_pending()
    time.sleep(1)

并行执行

默认情况下，Schedule 按顺序执行所有作业。其背后的原因是，很难找到让每个人都高兴的并行执行模型。

不过你可以通过多线程的形式来运行每个作业以解决此限制：

# Python 实用宝典
import threading
import time
import schedule

def 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)

日志记录

Schedule 模块同时也支持 logging 日志记录，这么使用：

# Python 实用宝典
import schedule
import logging

logging.basicConfig()
schedule_logger = logging.getLogger('schedule')
# 日志级别为DEBUG
schedule_logger.setLevel(level=logging.DEBUG)

def job():
    print("Hello, Logs")

schedule.every().second.do(job)

schedule.run_all()

schedule.clear()

效果如下：

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

异常处理

Schedule 不会自动捕捉异常，它遇到异常会直接抛出，这会导致一个严重的问题：后续所有的作业都会被中断执行，因此我们需要捕捉到这些异常。

你可以手动捕捉，但是某些你预料不到的情况需要程序进行自动捕获，加一个装饰器就能做到了：

# Python 实用宝典
import functools

def 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 / 0

schedule.every(5).minutes.do(bad_task)

这样，​<strong>​bad_task​</strong>​ 在执行时遇到的任何错误，都会被 ​<strong>​catch_exceptions ​</strong>​ 捕获，这点在保证调度任务正常运转的时候非常关键。

Atas ialah kandungan terperinci Sangat praktikal! Modul jadual, artifak tugas berkala Python!. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!