简介
.. contents::
croniter 为带有类似 cron 格式的 datetime 对象提供迭代功能。
::
_ _
___ _ __ ___ _ __ (_) |_ ___ _ __
/ __| '__/ _ \| '_ \| | __/ _ \ '__|
| (__| | | (_) | | | | | || __/ |
\___|_| \___/|_| |_|_|\__\___|_|
网站:https://github.com/kiorky/croniter
构建徽章
.. image:: https://github.com/kiorky/croniter/actions/workflows/cicd.yml/badge.svg :target: https://github.com/kiorky/croniter/actions/workflows/cicd.yml
使用方法
一个简单的例子::
>>> from croniter import croniter
>>> from datetime import datetime
>>> base = datetime(2010, 1, 25, 4, 46)
>>> iter = croniter('*/5 * * * *', base) # 每5分钟
>>> print(iter.get_next(datetime)) # 2010-01-25 04:50:00
>>> print(iter.get_next(datetime)) # 2010-01-25 04:55:00
>>> print(iter.get_next(datetime)) # 2010-01-25 05:00:00
>>>
>>> iter = croniter('2 4 * * mon,fri', base) # 每周一和周五的04:02
>>> print(iter.get_next(datetime)) # 2010-01-26 04:02:00
>>> print(iter.get_next(datetime)) # 2010-01-30 04:02:00
>>> print(iter.get_next(datetime)) # 2010-02-02 04:02:00
>>>
>>> iter = croniter('2 4 1 * wed', base) # 每周三或每月1日的04:02
>>> print(iter.get_next(datetime)) # 2010-01-27 04:02:00
>>> print(iter.get_next(datetime)) # 2010-02-01 04:02:00
>>> print(iter.get_next(datetime)) # 2010-02-03 04:02:00
>>>
>>> iter = croniter('2 4 1 * wed', base, day_or=False) # 每月1日是周三时的04:02
>>> print(iter.get_next(datetime)) # 2010-09-01 04:02:00
>>> print(iter.get_next(datetime)) # 2010-12-01 04:02:00
>>> print(iter.get_next(datetime)) # 2011-06-01 04:02:00
>>>
>>> iter = croniter('0 0 * * sat#1,sun#2', base) # 每月第一个周六和第二个周日
>>> print(iter.get_next(datetime)) # 2010-02-06 00:00:00
>>>
>>> iter = croniter('0 0 * * 5#3,L5', base) # 每月第三个和最后一个周五
>>> print(iter.get_next(datetime)) # 2010-01-29 00:00:00
>>> print(iter.get_next(datetime)) # 2010-02-19 00:00:00
你只需要了解如何使用构造函数和 get_next
方法,这些方法的签名如下::
>>> def __init__(self, cron_format, start_time=time.time(), day_or=True)
croniter 从 start_time
开始按照 cron_format
进行迭代。
cron_format
的格式为 分 时 日 月 星期,你可以参考
http://en.wikipedia.org/wiki/Cron 获取更多详情。day_or
开关用于控制 croniter 如何处理 日 和 星期 条目。默认选项是 cron 行为,
使用 OR 连接这些值。如果将开关设置为 False,则使用 AND 连接这些值。
这种行为类似于 fcron,使你能够定义例如每月第二个周五执行的任务,方法是同时设置月中的日期和星期几。
::
>>> def get_next(self, ret_type=float)
get_next 根据 cron 表达式计算下一个值,并返回 ret_type
类型的对象。
ret_type
应该是 float
或 datetime
对象。
支持 get_prev
方法。(>= 0.2.0)::
>>> base = datetime(2010, 8, 25)
>>> itr = croniter('0 0 1 * *', base)
>>> print(itr.get_prev(datetime)) # 2010-08-01 00:00:00
>>> print(itr.get_prev(datetime)) # 2010-07-01 00:00:00
>>> print(itr.get_prev(datetime)) # 2010-06-01 00:00:00
你可以使用 is_valid
类方法验证你的 cron 表达式。(>= 0.3.18)::
>>> croniter.is_valid('0 0 1 * *') # True
>>> croniter.is_valid('0 wrong_value 1 * *') # False
关于 DST
请确保使用带有时区信息的 datetime 初始化你的 croniter 实例,以使其正常工作!
使用 pytz 的示例::
>>> import pytz
>>> tz = pytz.timezone("Europe/Paris")
>>> local_date = tz.localize(datetime(2017, 3, 26))
>>> val = croniter('0 0 * * *', local_date).get_next(datetime)
使用 python_dateutil 的示例::
>>> import dateutil.tz
>>> tz = dateutil.tz.gettz('Asia/Tokyo')
>>> local_date = datetime(2017, 3, 26, tzinfo=tz)
>>> val = croniter('0 0 * * *', local_date).get_next(datetime)
使用 Python 内置模块的示例::
>>> from datetime import datetime, timezone
>>> local_date = datetime(2017, 3, 26, tzinfo=timezone.utc)
>>> val = croniter('0 0 * * *', local_date).get_next(datetime)
关于秒级重复
Croniter 能够处理秒级重复的 crontab 格式,默认情况下秒是第 6 个字段::
>>> base = datetime(2012, 4, 6, 13, 26, 10)
>>> itr = croniter('* * * * * 15,25', base)
>>> itr.get_next(datetime) # 4/6 13:26:15
>>> itr.get_next(datetime) # 4/6 13:26:25
>>> itr.get_next(datetime) # 4/6 13:27:15
你也可以注意到,这个表达式将从开始的 datetime 每秒重复一次。
croniter('* * * * * *', local_date).get_next(datetime)
你也可以在第一个字段使用秒:
>>> itr = croniter('15,25 * * * * *', base, second_at_beginning=True)
关于年份
Croniter 也支持年份字段。
年份位于第七个字段,即在秒重复之后。
年份字段的范围是从1970年到2099年。
要忽略秒重复,只需将秒设置为 0
或任何其他常量:
>>> base = datetime(2012, 4, 6, 2, 6, 59)
>>> itr = croniter('0 0 1 1 * 0 2020/2', base)
>>> itr.get_next(datetime) # 2020年1月1日 0:0:0
>>> itr.get_next(datetime) # 2022年1月1日 0:0:0
>>> itr.get_next(datetime) # 2024年1月1日 0:0:0
支持 start_time 偏移
参见 https://github.com/kiorky/croniter/pull/76, 你可以设置 start_time=,然后将 expand_from_start_time=True,这样生成的时间将从 start_time 开始计算,而不是从日历日开始:
>>> from pprint import pprint
>>> iter = croniter('0 0 */7 * *', start_time=datetime(2024, 7, 11), expand_from_start_time=True);pprint([iter.get_next(datetime) for a in range(10)])
[datetime.datetime(2024, 7, 18, 0, 0),
datetime.datetime(2024, 7, 25, 0, 0),
datetime.datetime(2024, 8, 4, 0, 0),
datetime.datetime(2024, 8, 11, 0, 0),
datetime.datetime(2024, 8, 18, 0, 0),
datetime.datetime(2024, 8, 25, 0, 0),
datetime.datetime(2024, 9, 4, 0, 0),
datetime.datetime(2024, 9, 11, 0, 0),
datetime.datetime(2024, 9, 18, 0, 0),
datetime.datetime(2024, 9, 25, 0, 0)]
>>> # 而不是默认行为:
>>> iter = croniter('0 0 */7 * *', start_time=datetime(2024, 7, 11), expand_from_start_time=False);pprint([iter.get_next(datetime) for a in range(10)])
[datetime.datetime(2024, 7, 15, 0, 0),
datetime.datetime(2024, 7, 22, 0, 0),
datetime.datetime(2024, 7, 29, 0, 0),
datetime.datetime(2024, 8, 1, 0, 0),
datetime.datetime(2024, 8, 8, 0, 0),
datetime.datetime(2024, 8, 15, 0, 0),
datetime.datetime(2024, 8, 22, 0, 0),
datetime.datetime(2024, 8, 29, 0, 0),
datetime.datetime(2024, 9, 1, 0, 0),
datetime.datetime(2024, 9, 8, 0, 0)]
测试日期是否匹配 crontab
使用以下方法测试匹配(>=0.3.32版本):
>>> croniter.match("0 0 * * *", datetime(2019, 1, 14, 0, 0, 0, 0))
True
>>> croniter.match("0 0 * * *", datetime(2019, 1, 14, 0, 2, 0, 0))
False
>>>
>>> croniter.match("2 4 1 * wed", datetime(2019, 1, 1, 4, 2, 0, 0)) # 每周三或每月1日的04:02
True
>>> croniter.match("2 4 1 * wed", datetime(2019, 1, 1, 4, 2, 0, 0), day_or=False) # 每月1日是周三时的04:02
False
测试 crontab 是否在日期范围内匹配
使用以下方法测试范围匹配(>=2.0.3版本):
>>> croniter.match_range("0 0 * * *", datetime(2019, 1, 13, 0, 59, 0, 0), datetime(2019, 1, 14, 0, 1, 0, 0))
True
>>> croniter.match_range("0 0 * * *", datetime(2019, 1, 13, 0, 1, 0, 0), datetime(2019, 1, 13, 0, 59, 0, 0))
False
>>> croniter.match_range("2 4 1 * wed", datetime(2019, 1, 1, 3, 2, 0, 0), datetime(2019, 1, 1, 5, 1, 0, 0))
# 每周三或每月1日的04:02
True
>>> croniter.match_range("2 4 1 * wed", datetime(2019, 1, 1, 3, 2, 0, 0), datetime(2019, 1, 1, 5, 2, 0, 0), day_or=False)
# 每月1日是周三时的04:02
False
日期匹配之间的间隔
出于性能考虑,croniter 限制了尝试查找下一个匹配项时消耗的 CPU 周期数。
从 v0.3.35 版本开始,这种行为可通过 max_years_between_matches
参数进行配置,默认窗口已从1年增加到50年。
对于许多用例来说,默认设置应该足够了。
评估多个 cron 表达式或处理来自不受信任源或最终用户的 cron 表达式的应用程序应使用此参数。
迭代稀疏的 cron 表达式可能会导致 CPU 消耗增加或引发 CroniterBadDateError
异常,表明 croniter 已放弃尝试查找下一个(或上一个)匹配项。
明确指定 max_years_between_matches
提供了一种限制 CPU 使用的方法,并通过消除对 CroniterBadDateError
的需求来简化可迭代接口。
可迭代接口的差异基于这样的推理:只要明确约定了 max_years_between_matches
,就不需要 croniter 发出信号表明它已放弃;简单地停止迭代更可取。
这个示例匹配1月1日星期五凌晨4点。
由于1月1日并不经常是星期五,因此每次出现之间可能会相隔几年。
将限制设置为15年可确保匹配所有结果:
>>> it = croniter("0 4 1 1 fri", datetime(2000,1,1), day_or=False, max_years_between_matches=15).all_next(datetime)
>>> for i in range(5):
... print(next(it))
...
2010-01-01 04:00:00
2016-01-01 04:00:00
2021-01-01 04:00:00
2027-01-01 04:00:00
2038-01-01 04:00:00
然而,如果只关心未来5年内的日期,只需在上面的示例中设置 max_years_between_matches=5
。
这将导致没有找到匹配项,但不会浪费额外的循环去寻找远未来的不需要的匹配项。
使用cron遍历范围
使用 croniter_range()
函数在范围内查找匹配项。这与内置的 range(start,stop,step)
函数非常相似,但适用于日期。step
参数是一个cron表达式。
在(>=0.3.34)版本中添加
列出2019年每月的第一个星期六:
>>> from croniter import croniter_range
>>> for dt in croniter_range(datetime(2019, 1, 1), datetime(2019, 12, 31), "0 0 * * sat#1"):
>>> print(dt)
哈希表达式
croniter支持Jenkins风格的哈希表达式,使用"H"定义关键字和必需的hash_id关键字参数。 给定相同的hash_id,哈希表达式保持一致,但不同的hash_id会产生完全不同的评估结果。 这允许,例如,在不需要手动分散的情况下,均匀分布具有不同名称的作业。
>>> itr = croniter("H H * * *", hash_id="hello")
>>> itr.get_next(datetime)
datetime.datetime(2021, 4, 10, 11, 10)
>>> itr.get_next(datetime)
datetime.datetime(2021, 4, 11, 11, 10)
>>> itr = croniter("H H * * *", hash_id="hello")
>>> itr.get_next(datetime)
datetime.datetime(2021, 4, 10, 11, 10)
>>> itr = croniter("H H * * *", hash_id="bonjour")
>>> itr.get_next(datetime)
datetime.datetime(2021, 4, 10, 20, 52)
随机表达式
支持随机"R"定义关键字,它们仅在其croniter()实例内保持一致。
>>> itr = croniter("R R * * *")
>>> itr.get_next(datetime)
datetime.datetime(2021, 4, 10, 22, 56)
>>> itr.get_next(datetime)
datetime.datetime(2021, 4, 11, 22, 56)
>>> itr = croniter("R R * * *")
>>> itr.get_next(datetime)
datetime.datetime(2021, 4, 11, 4, 19)
关键字表达式
支持Vixie cron风格的"@"关键字表达式。 它们的评估结果取决于是否提供hash_id:没有hash_id对应Vixie cron定义(精确时间,分钟分辨率),而有hash_id对应Jenkins定义(在周期内哈希,秒分辨率)。
============ ============ ================
关键字 无hash_id 有hash_id
============ ============ ================
@midnight 0 0 * * * H H(0-2) * * * H
@hourly 0 * * * * H * * * * H
@daily 0 0 * * * H H * * * H
@weekly 0 0 * * 0 H H * * H H
@monthly 0 0 1 * * H H H * * H
@yearly 0 0 1 1 * H H H H * H
@annually 0 0 1 1 * H H H H * H
============ ============ ================
升级
至2.0.0版本
- 如果您安装的pytz版本
<=2021.1
,请使用requirements/base.txt中指定的版本安装或升级pytz。
开发此包
::
git clone https://github.com/kiorky/croniter.git
cd croniter
virtualenv --no-site-packages venv
. venv/bin/activate
pip install --upgrade -r requirements/test.txt
py.test src
发布新版本
我们使用zest.fullreleaser,这是一个优秀的发布基础设施。
执行并遵循以下指令 ::
. venv/bin/activate
pip install --upgrade -r requirements/release.txt
./release.sh
贡献者
感谢所有为这个项目做出贡献的人! 如果您做出了贡献但您的名字没有列在下面,请告诉我。
- mrmachine
- Hinnack
- shazow
- kiorky
- jlsandell
- mag009
- djmitche
- GreatCombinator
- chris-baynes
- ipartola
- yuzawa-san
- lowell80 (Kintyre)
- scop
- zed2015
- Ryan Finnie (rfinnie)
- salitaba