Python Schedule --定时任务调度

发表于 分类于

schedule 用于周期性执行 Python 函数，可以设定天、周、时、分、秒计划运行函数

安装

1
pip3 install schedule -i https://pypi.tuna.tsinghua.edu.cn/simple

方法

every()

  • every(interval: int = 1)
  • 安排一个新的定期工作

每2s执行

1
every(2).seconds.do(job)

每10分钟执行

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

每小时执行

1
every().hour.do(job)

每天 10:30:01 执行

1
every().day.at("10:30:01").do(job)

每星期几的 13:15:01 执行

1
2
3
4
5
6
7
every().monday.at("13:15:01").do(job)      # 星期一
every().tuesday.at("13:15:01").do(job)     # 星期二
every().wednesday.at("13:15:01").do(job)   # 星期三
every().thursday.at("13:15:01").do(job)    # 星期四
every().friday.at("13:15:01").do(job)      # 星期五
every().saturday.at("13:15:01").do(job)    # 星期六
every().sunday.at("13:15:01").do(job)      # 星期天

每隔5到10天执行一次job()函数

1
every(5).to(10).days.do(job)

一个例子

1
2
3
4
5
6
7
8
9
10
11
12
13
from schedule import every, run_pending


def job_1():
    print("job_1 working...")


# 创建一个每2s执行的调度任务
every(2).seconds.do(job_1)


while True:
    run_pending()   # 运行所有可运行的任务

输出

1
2
3
4
5
job_1 working...
job_1 working...
job_1 working...
...
...

repeat()

  • repeat(job, *args, **kwargs)
  • 装饰函数来安排一个新的定期工作
  • 任何额外的参数都会在作业运行时传递给装饰的函数

每2s执行

1
@repeat(every(2).seconds)

每10分钟执行

1
@repeat(every(10).minutes)

每小时执行

1
@repeat(every().hour)

每天 10:30:01 执行

1
@repeat(every().day.at("10:30:01"))

每星期几的 13:15:01 执行

1
2
3
4
5
6
7
@repeat(every().monday.at("13:15:01"))     # 星期一
@repeat(every().tuesday.at("13:15:01"))    # 星期二
@repeat(every().wednesday.at("13:15:01"))  # 星期三
@repeat(every().thursday.at("13:15:01"))   # 星期四
@repeat(every().friday.at("13:15:01"))     # 星期五
@repeat(every().saturday.at("13:15:01"))   # 星期六
@repeat(every().sunday.at("13:15:01"))     # 星期天

每隔5到10天执行一次job()函数

1
@repeat(every(5).to(10).days)

一个例子

1
2
3
4
5
6
7
8
9
10
11
from schedule import every, repeat, run_pending


# 使用装饰器创建一个每2s执行的调度任务
@repeat(every(2).seconds)
def job_2():
    print("job_2 working...")


while True:
    run_pending()   # 运行所有可运行的任务

输出

1
2
3
4
5
job_2 working...
job_2 working...
job_2 working...
...
...

run_pending()

  • run_pending()
  • 运行所有计划中的工作
  • 请注意，run_pending()的预期行为是不运行错过的作业
  • 比如说，如果你注册了一个应该每分钟运行的作业，而你只在一小时内调用run_pending()，那么你的作业就不会在中间运行60次，而只会运行一次

一个例子

1
2
3
4
5
6
7
8
9
10
11
12
13
from schedule import every, run_pending


def job_1():
    print("job_1 working...")


# 创建一个每2s执行的调度任务
every(2).seconds.do(job_1)


while True:
    run_pending()   # 运行所有可运行的任务

输出

1
2
3
4
5
job_1 working...
job_1 working...
job_1 working...
...
...

run_all()

  • run_all(delay_seconds: int = 0)
  • 运行所有工作，不管它们是否被安排运行
  • 每个作业之间会有一个 delay 秒的延迟。这有助于将作业产生的系统负载更均匀地分布在一段时间内

一个例子

1
2
3
4
5
6
7
8
9
10
11
12
from schedule import every, run_all


def job_1():
    print("job_1 working...")


# 创建一个每2s执行的调度任务
every(2).seconds.do(job_1)

run_all(2)  # 延迟2秒后运行所有工作
run_all()   # 运行所有工作

输出

1
2
job_1 working...
job_1 working...

get_jobs()

  • get_jobs(tag: Optional[Hashable] = None)
  • 获取标有给定标签的计划工作，如果省略标签，则获取所有工作

一个例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from schedule import every, get_jobs


def job_1():
    print("job_1 working...")


def job_2():
    print("job_2 working...")


# 创建一个每5s执行的调度任务
every(5).seconds.do(job_1)

# 创建一个每20s执行的调度任务并设置标签
every(20).seconds.do(job_2).tag('秒', 'seconds')

print(get_jobs('秒'))       # 获取带有 秒 标签的工作 
print(get_jobs('seconds'))  # 获取带有 seconds 标签的工作 
print(get_jobs())           # 获取所有工作

输出

1
2
3
[Every 20 seconds do job_2() (last run: [never], next run: 2022-04-18 22:44:24)]
[Every 20 seconds do job_2() (last run: [never], next run: 2022-04-18 22:44:24)]
[Every 5 seconds do job_1() (last run: [never], next run: 2022-04-18 22:44:09), Every 20 seconds do job_2() (last run: [never], next run: 2022-04-18 22:44:24)]

clear()

  • clear(tag: Optional[Hashable] = None)
  • 删除标有给定标签的计划工作，如果省略标签，则删除所有工作

一个例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from schedule import every, clear, get_jobs


def job_1():
    print("job_1 working...")


def job_2():
    print("job_2 working...")


# 创建一个每5s执行的调度任务
every(5).seconds.do(job_1)

# 创建一个每20s执行的调度任务并设置标签
every(20).seconds.do(job_2).tag('秒', 'seconds')

print(get_jobs())   # 获取所有工作
print(clear('秒'))  # 删除带有 秒 标签的工作 
print(get_jobs())   # 获取所有工作
print(clear())      # 删除所有工作
print(get_jobs())   # 获取所有工作

输出

1
2
3
4
5
[Every 5 seconds do job_1() (last run: [never], next run: 2022-04-18 22:50:07), Every 20 seconds do job_2() (last run: [never], next run: 2022-04-18 22:50:22)]
None
[Every 5 seconds do job_1() (last run: [never], next run: 2022-04-18 22:50:07)]
None
[]

cancel_job()

  • cancel_job(job: Job)
  • 删除一个预定的工作

一个例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from schedule import every, cancel_job, get_jobs


def job_1():
    print("job_1 working...")


def job_2():
    print("job_2 working...")


# 创建一个每5s执行的调度任务
every(5).seconds.do(job_1)

# 创建一个每20s执行的调度任务并设置标签
every(20).seconds.do(job_2).tag('秒', 'seconds')

print(get_jobs())           # 获取所有工作
print(cancel_job(job_1))    # 删除 job_1 工作
print(get_jobs())           # 获取所有工作

输出

1
2
3
[Every 5 seconds do job_1() (last run: [never], next run: 2022-04-18 22:54:09), Every 20 seconds do job_2() (last run: [never], next run: 2022-04-18 22:54:24)]
None
[Every 5 seconds do job_1() (last run: [never], next run: 2022-04-18 22:54:09), Every 20 seconds do job_2() (last run: [never], next run: 2022-04-18 22:54:24)]

next_run()

  • next_run()
  • 下一个作业应该运行的日期

一个例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import time
from schedule import every, next_run, run_pending


def job_1():
    print("job_1 working...")


# 创建一个每3s执行的调度任务
every(3).seconds.do(job_1)


while True:
    run_pending()        # 运行所有可运行的任务
    print(next_run())    # 获取下一个作业应该运行的日期
    time.sleep(1)

输出

1
2
3
4
5
6
7
8
2022-04-18 23:00:20.936598
2022-04-18 23:00:20.936598
2022-04-18 23:00:20.936598
job_1 working...
2022-04-18 23:00:23.960987
2022-04-18 23:00:23.960987
...
...

idle_seconds()

  • idle_seconds()
  • 还有多少秒即将开始运行任务

一个例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import time
from schedule import every, idle_seconds, run_pending


def job_1():
    print("job_1 working...")


# 创建一个每3s执行的调度任务
every(3).seconds.do(job_1)


while True:
    run_pending()           # 运行所有可运行的任务
    print(idle_seconds())   # 还有多少秒即将开始运行任务
    time.sleep(1)

输出

1
2
3
4
5
6
7
8
9
3.0
1.989047
0.974167
job_1 working...
3.0
1.995853
0.98298
...
...

资料