Validus A dead simple Python data validation library.…
December 30, 2019
Python 自动任务, Python 定时任务, parse-crontab, Parse and use crontab schedules in Python
项目地址:https://github.com/josiahcarlson/parse-crontab
描述
该软件包旨在提供一种解析crontab计划条目并确定下一次应何时运行项目的方法。更具体地说,它计算从调用.next()方法到下一次应执行该项目之间的延迟(以秒为单位)。
将下面的图表与http://en.wikipedia.org/wiki/Cron#CRON_expression进行比较, 您会注意到不支持W和#符号。
Field Name 栏位名称 | Mandatory 强制性的 | Allowed Values 允许值 | Default Value 默认值 | Allowed Special Characters 允许的特殊字符 |
---|---|---|---|---|
Seconds 秒 | No | 0-59 | 0 | * / , – |
Minutes 分钟 | Yes | 0-59 | N/A | * / , – |
Hours 小时 | Yes | 0-23 | N/A | * / , – |
Day of month 一个月中的某天 | Yes | 1-31 | N/A | * / , – ? L |
Month 月 | Yes | 1-12 或 JAN-DEC | N/A | * / , – |
Day of week 星期几 | Yes | 0-6 或 SUN-SAT | N/A | * / , – ? L |
Year 年 | No | 1970-2099 |
|
* / , – |
如果您的cron条目有5个值,则使用“星期几”-“星期几”,默认秒数为“秒”,并且默认年份为“年”。如果您的cron条目具有6个值,则使用分钟-年,并且默认秒为前缀。
因此,仅接受5-7个值的crontab条目(并根据需要改成7个值)。
样本单个crontab字段
受支持的条目的示例如下
* */5 7/8 3-25/7 3,7,9 0-10,30-40/5
对于月份或星期几,可以在任何可选的/的左侧使用月份或日期的3个字母缩写。
一周中的几天:
mon-fri sun-thu/2
对于月份:
apr-jul mar-sep/3
安装
pip install crontab
示例用途
>>> from crontab import CronTab >>> from datetime import datetime >>> # define the crontab for 25 minutes past the hour every hour ... entry = CronTab('25 * * * *') >>> # find the delay from when this was run (around 11:13AM) ... entry.next() 720.81637899999998 >>> # find the delay from when it was last scheduled ... entry.next(datetime(2011, 7, 17, 11, 25)) 3600.0
笔记
“一周中的某天”或“一个月中的某天”最多可以是“?”以外的值。要么 ‘*’。在指定了这些值之一的情况下,我们违反了此处的规范,并允许’*’作为’?’的别名(参见某些平台不支持’?’)。
该模块还支持方便的别名:
@yearly @annually @monthly @weekly @daily @hourly
完整的crontab条目示例及其含义:
30 */2 * * * -> 30 minutes past the hour every 2 hours 15,45 23 * * * -> 11:15PM and 11:45PM every day 0 1 ? * SUN -> 1AM every Sunday 0 1 * * SUN -> 1AM every Sunday (same as above) 0 0 1 jan/2 * 2011-2013 -> midnight on January 1, 2011 and the first of every odd month until the end of 2013 24 7 L * * -> 7:24 AM on the last day of every month 24 7 * * L5 -> 7:24 AM on the last friday of every month 24 7 * * Lwed-fri -> 7:24 AM on the last wednesday, thursday, and friday of every month
test_crontab.py
from collections import namedtuple import datetime import unittest import pytz import dateutil.tz from crontab import CronTab Results = namedtuple('Results', 'crontab delay max_delay now future') class TestCrontab(unittest.TestCase): def _run_test(self, crontab, max_delay, now=None, min_delay=None): ct = CronTab(crontab) now = now or datetime.datetime.utcnow() delay = ct.next(now, default_utc=True) assert delay is not None dd = Results(crontab, delay, max_delay, now, now+datetime.timedelta(seconds=delay)) assert delay <= max_delay, dd if min_delay is not None: assert delay >= min_delay, dd if not crontab.endswith(' 2099'): delay2 = ct.previous(now + datetime.timedelta(seconds=delay), default_utc=True) dd = Results(crontab, delay, max_delay, now, now+datetime.timedelta(seconds=delay)) assert abs(delay2) >= delay, (delay, delay2) pt = now + datetime.timedelta(seconds=delay) + datetime.timedelta(seconds=delay2) assert pt <= now, dd def _run_impossible(self, crontab, now): ct = CronTab(crontab) delay = ct.next(now, default_utc=True) assert delay is None, (crontab, delay, now, now+datetime.timedelta(seconds=delay)) def test_closest(self): ce = CronTab("*/15 10-15 * * 1-5") t945 = datetime.datetime(2013, 1, 1, 9, 45) # tuesday t1245 = datetime.datetime(2013, 1, 1, 12, 45) # tuesday s1245 = datetime.datetime(2013, 1, 5, 12, 45) # saturday assert not ce.test(t945) assert not ce.test(s1245) assert ce.test(t1245) n = datetime.datetime.utcfromtimestamp(ce.next(t945, delta=False, default_utc=True)) assert n == datetime.datetime(2013, 1, 1, 10, 0), n p = datetime.datetime.utcfromtimestamp(ce.previous(t945, delta=False, default_utc=True)) assert p == datetime.datetime(2012, 12, 31, 15, 45), p n = datetime.datetime.utcfromtimestamp(ce.next(s1245, delta=False, default_utc=True)) assert n == datetime.datetime(2013, 1, 7, 10, 0), n p = datetime.datetime.utcfromtimestamp(ce.previous(s1245, delta=False, default_utc=True)) assert p == datetime.datetime(2013, 1, 4, 15, 45), p def test_normal(self): self._run_test('*/5 * * * * * *', 5) self._run_test('* * * * *', 60) self._run_test('0 * * * *', 3600) self._run_test('0 0 * * *', 86400) self._run_test('0 0 1 * *', 31*86400) self._run_test('5/15 * * * *', 15*60) self._run_test('5-51/15 * * * *', 15*60) self._run_test('1,8,40 * * * *', 3600) self._run_test('0 0 1 1 *', 86400 * 366) self._run_test('0 0 1 1 * 2099', 99 * 86400 * 366) self._run_test('0 0 ? * 0-6', 86400) self._run_test('0 0 31 * *', 62*86400, datetime.datetime(2011, 1, 31)) self._run_test('0,1/2 * * * *', 60, datetime.datetime(2011, 1, 1, 1, 0)) self._run_test('0,1/2 * * * *', 120, datetime.datetime(2011, 1, 1, 1, 1)) self._run_test('0,1/2 * * * *', 60, datetime.datetime(2011, 1, 1, 1, 2)) self._run_test('0-6,50-59/2 * * * *', 60, datetime.datetime(2011, 1, 1, 1, 1)) self._run_test('0-6,50-59/2 * * * *', 120, datetime.datetime(2011, 1, 1, 1, 2)) self._run_test('0-6,50-59/2 * * * *', 900, datetime.datetime(2011, 1, 1, 1, 45)) self._run_test('0-6,50-59/2 * * * *', 60, datetime.datetime(2011, 1, 1, 1, 55)) self._run_test('0-6,50/2 * * * *', 60, datetime.datetime(2011, 1, 1, 1, 55)) self._run_test('10,20 15 * * *', 9*60, datetime.datetime(2011, 1, 1, 15, 1), min_delay=9*60) self._run_test('10,20 15 * * *', 5*60, datetime.datetime(2011, 1, 1, 15, 15), min_delay=5*60) self._run_test('10,20 15 * * *', 86400 - 600, datetime.datetime(2011, 1, 1, 15, 20), min_delay=86400 - 600) self._run_test('* 2-5 * * *', 12525, datetime.datetime(2013, 6, 19, 22, 31, 15), min_delay=12525) def test_alternate(self): self._run_test('0 0 1 jan-dec *', 32 * 86400) self._run_test('0 0 ? * sun-sat', 86400) def test_aliases(self): self._run_test('@hourly', 3600) self._run_test('@daily', 86400) self._run_test('@monthly', 31*86400) self._run_test('@yearly', 86400 * 366) def test_day_of_week(self): self._run_test('0 0 ? 7 mon', 4*86400, datetime.datetime(2011, 7, 15)) self._run_test('0 0 ? 7 mon', 366*86400, datetime.datetime(2011, 7, 25, 1)) self._run_test('0 0 ? 8 mon-fri', 5*86400 + 1, datetime.datetime(2011, 7, 27, 1)) self._run_test('0 12 * * sat-sun', 129600, datetime.datetime(2015, 11, 6), 129600) self._run_test('0 12 * * sat-sun', 86400, datetime.datetime(2015, 11, 7, 12), 86400) self._run_test('0 12 * * sat-sun', 518400, datetime.datetime(2015, 11, 8, 12), 518400) self._run_test('0 5 * * fri *', 7*86400, datetime.datetime(2016, 3, 25, 5), 7*86400) self._run_test('* * * * Fri *', 6*86400+1, datetime.datetime(2016, 3, 25, 23, 59, 59), 6*86400+1) self._run_test('* * * * Fri *', 60, datetime.datetime(2016, 3, 25, 23, 56), 60) self._run_test('* * 13 * Fri *', 181*86400+1, datetime.datetime(2015, 11, 13, 23, 59, 59), 181*86400+1) def test_last_day(self): self._run_test('0 0 L 2 ?', 28*86400, datetime.datetime(2011, 1, 31)) self._run_test('0 0 1,L 2 ?', 86400, datetime.datetime(2011, 1, 31)) self._run_test('0 0 2,L 2 ?', 2*86400, datetime.datetime(2011, 1, 31)) self._run_test('0 0 L 2 ?', 58*86400, datetime.datetime(2011, 1, 1)) self._run_test('0 0 ? 2 L1', 58*86400, datetime.datetime(2011, 1, 31)) self._run_test('0 0 ? 7 L1', 1*86400, datetime.datetime(2011, 7, 24)) self._run_test('0 0 ? 7 L2', 2*86400, datetime.datetime(2011, 7, 24)) self._run_test('0 0 ? 7 L3', 3*86400, datetime.datetime(2011, 7, 24)) self._run_test('0 0 ? 7 L4', 4*86400, datetime.datetime(2011, 7, 24)) self._run_test('0 0 ? 7 L5', 5*86400, datetime.datetime(2011, 7, 24)) self._run_test('0 0 ? 7 L6', 6*86400, datetime.datetime(2011, 7, 24)) self._run_test('0 0 ? 7 L0', 7*86400, datetime.datetime(2011, 7, 24)) self._run_test('0 0 ? 7 L3-5', 3*86400, datetime.datetime(2011, 7, 24)) self._run_test('0 0 ? 7 L3-5', 2*86400, datetime.datetime(2011, 7, 25)) self._run_test('0 0 ? 7 L3-5', 86400, datetime.datetime(2011, 7, 26)) self._run_test('0 0 ? 7 L3-5', 86400, datetime.datetime(2011, 7, 27)) self._run_test('0 0 ? 7 L3-5', 86400, datetime.datetime(2011, 7, 28)) self._run_test('0 0 ? 7 L3-5', 362*86400, datetime.datetime(2011, 7, 29)) self._run_test('0 0 ? 7 L0-1', 24*86400, datetime.datetime(2011, 7, 1)) self._run_test('0 0 ? 7 L0-1', 24*86400, datetime.datetime(2011, 7, 1)) self._run_test('0 0 ? 7 L0-1', 86400, datetime.datetime(2011, 7, 24)) self._run_test('0 0 ? 7 L0-1', 6*86400, datetime.datetime(2011, 7, 25)) self._run_test('59 23 L 12 *', 282*86400, datetime.datetime(2012, 3, 25), 280*84400) self._run_test('0 0 ? 2 L1', 28*86400, datetime.datetime(2016, 2, 1), 28*86400) self._run_test('0 0 ? 2 L0', 27*86400, datetime.datetime(2016, 2, 1), 27*86400) self._run_test('0 0 ? 2 L7', 27*86400, datetime.datetime(2016, 2, 1), 27*86400) self._run_test('0 0 ? 2 L6', 26*86400, datetime.datetime(2016, 2, 1), 26*86400) self._run_test('0 0 ? 2 L5', 25*86400, datetime.datetime(2016, 2, 1), 25*86400) self._run_test('0 0 ? 2 L4', 24*86400, datetime.datetime(2016, 2, 1), 24*86400) self._run_test('0 0 ? 2 L3', 23*86400, datetime.datetime(2016, 2, 1), 23*86400) self._run_test('0 0 ? 2 L2', 22*86400, datetime.datetime(2016, 2, 1), 22*86400) def test_impossible(self): self._run_impossible('0 0 * 7 fri 2011', datetime.datetime(2011, 7, 31)) self._run_impossible('0 0 29 2 * 2011', datetime.datetime(2011, 2, 1)) self._run_impossible('0 0 L 1 * 2011', datetime.datetime(2011, 2, 1)) self._run_impossible('0 0 L 1 * 2011', datetime.datetime(2011, 1, 31)) self._run_impossible('0 0 ? 7 L3-5 2011', datetime.datetime(2011, 7, 29)) self._run_impossible('0 0 29 2 * 2012-2015', datetime.datetime(2012, 2, 29)) def test_bad_crontabs(self): self.assertRaises(ValueError, lambda: CronTab('*')) self.assertRaises(ValueError, lambda: CronTab('* *')) self.assertRaises(ValueError, lambda: CronTab('* * *')) self.assertRaises(ValueError, lambda: CronTab('* * * *')) self.assertRaises(ValueError, lambda: CronTab('* * * * * * * *')) self.assertRaises(ValueError, lambda: CronTab('-1 * * * *')) self.assertRaises(ValueError, lambda: CronTab('* mon-tue * * *')) self.assertRaises(ValueError, lambda: CronTab('* * * feb-jan *')) self.assertRaises(ValueError, lambda: CronTab('* * * * L')) self.assertRaises(ValueError, lambda: CronTab('* * * L *')) self.assertRaises(ValueError, lambda: CronTab('* L * * *')) self.assertRaises(ValueError, lambda: CronTab('L * * * *')) self.assertRaises(ValueError, lambda: CronTab('* 1, * * *')) self.assertRaises(ValueError, lambda: CronTab('60 * * * *')) self.assertRaises(ValueError, lambda: CronTab('* 25 * * *')) self.assertRaises(ValueError, lambda: CronTab('* * 32 * *')) self.assertRaises(ValueError, lambda: CronTab('* * * 13 *')) self.assertRaises(ValueError, lambda: CronTab('* * * * 9999')) self.assertRaises(ValueError, lambda: CronTab('20/50 * * * * *')) self.assertRaises(ValueError, lambda: CronTab('0/100 * * * * *')) self.assertRaises(ValueError, lambda: CronTab('*,50-59/12 * * * *')) self.assertRaises(ValueError, lambda: CronTab('* * * DEC/7 * *')) self.assertRaises(ValueError, lambda: CronTab('* * * * MON/7 *')) def test_previous(self): schedule = CronTab('0 * * * *') ts = datetime.datetime(2014, 6, 6, 9, 0, 0) for i in range(70): next = schedule.next(ts, default_utc=True) self.assertTrue(0 <= next <= 3600, next) previous = schedule.previous(ts, default_utc=True) self.assertTrue(-3600 <= previous <= 0, previous) ts += datetime.timedelta(seconds=1) def test_timezones_pytz(self): s = CronTab('0 9 13 3 * 2016') self.assertEqual(s.next(datetime.datetime(2016, 3, 13), default_utc=True), 32400) self.assertEqual(s.next(pytz.utc.localize(datetime.datetime(2016, 3, 13)), default_utc=True), 32400) self.assertEqual(s.next(pytz.timezone('US/Eastern').localize(datetime.datetime(2016, 3, 13))), 28800) t = CronTab('0 9 * * * 2018') self.assertEqual(t.next(datetime.datetime(2018, 11, 4), default_utc=True), 32400) timezone = pytz.timezone("America/Los_Angeles") self.assertEqual(t.next(timezone.localize(datetime.datetime(2018, 11, 4))), 36000) before = pytz.utc.localize(datetime.datetime(2018, 11, 4, 8, 29)).astimezone(timezone) self.assertEqual(CronTab('30 1 * * * 2018').next(before), 3660) self.assertEqual(CronTab('30 1 * * * 2018').next(timezone.localize(datetime.datetime(2018, 11, 4, 1, 15))), 900) self.assertEqual(CronTab('30 1 * * * 2018').next(timezone.localize(datetime.datetime(2018, 11, 4))), 9000) def test_timezones_dateutil(self): s = CronTab('0 9 13 3 * 2016') self.assertEqual(s.next(datetime.datetime(2016, 3, 13), default_utc=True), 32400) utc = dateutil.tz.tzutc() self.assertEqual(s.next(datetime.datetime(2016, 3, 13, tzinfo=utc), default_utc=True), 32400) x = datetime.datetime(2016, 3, 13, tzinfo=dateutil.tz.gettz('US/Eastern')) self.assertEqual(s.next(x), 28800) t = CronTab('0 9 * * * 2018') self.assertEqual(t.next(datetime.datetime(2018, 11, 4), default_utc=True), 32400) timezone = dateutil.tz.gettz("America/Los_Angeles") self.assertEqual(t.next(datetime.datetime(2018, 11, 4, tzinfo=timezone)), 36000) before = datetime.datetime(2018, 11, 4, 8, 29, tzinfo=utc).astimezone(timezone) self.assertEqual(CronTab('30 1 * * * 2018').next(before), 60) before = datetime.datetime(2018, 11, 4, 8, 31, tzinfo=utc).astimezone(timezone) self.assertEqual(CronTab('30 1 * * * 2018').next(before), 89940) self.assertEqual(CronTab('30 1 * * * 2018').next(datetime.datetime(2018, 11, 4, 1, 15, tzinfo=timezone)), 900) self.assertEqual(CronTab('30 1 * * * 2018').next(datetime.datetime(2018, 11, 4, tzinfo=timezone)), 5400) if __name__ == '__main__': unittest.main()
本文:Python 自动任务, Python 定时任务, parse-crontab, Parse and use crontab schedules in Python