kumi/parse_crontab
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Add code, tests, makefile, etc.

Browse source
This commit is contained in:
Josiah Carlson 2011-07-17 17:28:33 -07:00
parent cca54a6f5f
commit 3e2ef51072
8 changed files with 501 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -1 +1,4 @@
*.pyc
MANIFEST
build/*
dist/*

14
Makefile Normal file
View file

@ -0,0 +1,14 @@
SHELL=/bin/bash
clean:
-rm -f *.pyc crontab/*.pyc README.html MANIFEST
-rm -rf build dist
install:
python setup.py install
test:
python -m tests.test_crontab
upload:
python setup.py sdist upload

96
README Normal file
View file

@ -0,0 +1,96 @@
Description
===========
This package intends to offer a method of parsing crontab schedule entries and
determining when an item should next be run. More specifically, it calculates
a delay in seconds from when the .next() method is called to when the item
should next be executed.
Comparing the below chart to http://en.wikipedia.org/wiki/Cron#CRON_expression
you will note that W and # symbols are not supported.
============= =========== ================= ===========================
Field Name Mandatory Allowed Values Allowed Special Characters
============= =========== ================= ===========================
Minutes Yes 0-59 \* / , -
Hours Yes 0-23 \* / , -
Day of month Yes 1-31 \* / , - ? L
Month Yes 1-12 or JAN-DEC \* / , -
Day of week Yes 0-6 or SUN-SAT \* / , - ? L
Year No 1970-2099 \* / , -
============= =========== ================= ===========================
Sample individual crontab fields
================================
Examples of supported entries are as follows::
*
*/5
7/8
3-25/7
3,7,9
0-10,30-40/5
For month or day of week entries, 3 letter abbreviations of the month or day
can be used to the left of any optional / where a number could be used.
For days of the week::
mon-fri
sun-thu/2
For month::
apr-jul
mar-sep/3
Example uses
============
::
>>> 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
Notes
=====
At most one of 'day of week' or 'day of month' can be a value other than '?'
or '*'. We violate spec here and allow '*' to be an alias for '?', in the case
where one of those values is specified (seeing as some platforms don't support
'?').
This module also supports the convenient aliases::
@yearly
@annually
@monthly
@weekly
@daily
@hourly
Example full crontab entries and their meanings::
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

4
crontab/__init__.py Normal file
View file

@ -0,0 +1,4 @@
from ._crontab import CronTab
__all__ = ['CronTab']

255
crontab/_crontab.py Normal file
View file

@ -0,0 +1,255 @@
'''
crontab.py
Written July 15, 2011 by Josiah Carlson
Released under the GNU GPL v2
available: http://www.gnu.org/licenses/gpl-2.0.html
Other licenses may be available upon request.
'''
from collections import namedtuple
import datetime
_ranges = [
(0, 59),
(0, 23),
(1, 31),
(1, 12),
(0, 6),
(1970, 2099),
]
_attribute = [
'minute',
'hour',
'day',
'month',
'isoweekday',
'year'
]
_alternate = {
3: {'jan': 1, 'feb': 2, 'mar': 3, 'apr': 4, 'may': 5, 'jun': 6,
'jul': 7, 'aug': 8, 'sep': 9, 'oct': 10, 'nov':11, 'dec':12},
4: {'sun': 0, 'mon': 1, 'tue': 2, 'wed': 3, 'thu': 4, 'fri': 5, 'sat': 6},
}
_aliases = {
'@yearly': '0 0 1 1 *',
'@annually': '0 0 1 1 *',
'@monthly': '0 0 1 * *',
'@weekly': '0 0 * * 0',
'@daily': '0 0 * * *',
'@hourly': '0 * * * *',
}
MINUTE = datetime.timedelta(minutes=1)
HOUR = datetime.timedelta(hours=1)
DAY = datetime.timedelta(days=1)
WEEK = datetime.timedelta(days=7)
MONTH = datetime.timedelta(days=28)
YEAR = datetime.timedelta(days=365)
def _day_incr(dt, m):
if m.day.input != 'l':
return DAY
odt = dt
ndt = dt = dt + DAY
while dt.month == ndt.month:
dt += DAY
return dt - odt
def _month_incr(dt, m):
odt = dt
dt += MONTH
while dt.month == odt.month:
dt += DAY
# get to the first of next month, let the backtracking handle it
dt = dt.replace(day=1)
return dt - odt
def _year_incr(dt, m):
# simple leapyear stuff works for 1970-2099 :)
mod = dt.year % 4
if mod == 0 and (dt.month, dt.day) < (2, 29):
return YEAR + DAY
if mod == 3 and (dt.month, dt.day) > (2, 29):
return YEAR + DAY
return YEAR
_increments = [
lambda *a: MINUTE,
lambda *a: HOUR,
_day_incr,
_month_incr,
lambda *a: DAY,
_year_incr,
]
Matcher = namedtuple('Matcher', 'minute, hour, day, month, weekday, year')
def _assert(condition, message, *args):
if not condition:
raise ValueError(message%args)
class _Matcher(object):
__slots__ = 'allowed', 'end', 'any', 'input', 'which'
def __init__(self, which, entry):
_assert(0 <= which <= 5,
"improper number of cron entries specified")
self.input = entry.lower()
self.which = which
self.allowed, self.end = self._parse_crontab(which, entry.lower())
self.any = self.allowed is None
def __call__(self, v, dt):
if self.input == 'l':
return v != (dt + DAY).month
elif self.input.startswith('l'):
okay = dt.month != (dt + WEEK).month
return okay and (self.any or v in self.allowed)
return self.any or v in self.allowed
def __lt__(self, other):
if self.any:
return self.end < other
return all(item < other for item in self.allowed)
def _parse_crontab(self, which, entry):
'''
This parses a single crontab field and returns the data necessary for
this matcher to accept the proper values.
See the README for information about what is accepted.
'''
# this handles day of week/month abbreviations
def _fix(it):
if which in _alternate and not it.isdigit():
if it in _alternate[which]:
return _alternate[which][it]
_assert(it.isdigit(),
"invalid range specifier: %r (%r)", it, entry)
return int(it, 10)
# this handles individual items/ranges
def _parse_piece(it):
if '-' in it:
start, end = map(_fix, it.split('-'))
elif it == '*':
start = _start
end = _end
else:
start = _fix(it)
end = _end
if increment is None:
return set([start])
_assert(_start <= start <= _end,
"range start value %r out of range [%r, %r]", start, _start, _end)
_assert(_start <= end <= _end,
"range end value %r out of range [%r, %r]", end, _start, _end)
_assert(start <= end,
"range start value %r > end value %r", start, end)
return set(range(start, end+1, increment or 1))
_start, _end = _ranges[which]
# wildcards
if entry in ('*', '?'):
if entry == '?':
_assert(which in (2, 4),
"cannot use '?' in the %r field", _attribute[which])
return None, _end
# last day of the month
if entry == 'l':
_assert(which == 2,
"you can only specify a bare 'L' in the 'day' field")
return None, _end
# last day of the week
elif entry.startswith('l'):
_assert(which == 4,
"you can only specify a leading 'L' in the 'weekday' field")
entry = entry.lstrip('l')
increment = None
# increments
if '/' in entry:
entry, increment = entry.split('/')
increment = int(increment, 10)
_assert(increment > 0,
"you can only use positive increment values, you provided %r",
increment)
# handle all of the a,b,c and x-y,a,b entries
good = set()
for it in entry.split(','):
good.update(_parse_piece(it))
return good, _end
class CronTab(object):
__slots__ = 'matchers',
def __init__(self, crontab):
self.matchers = self._make_matchers(crontab)
def _make_matchers(self, crontab):
'''
This constructs the full matcher struct.
'''
crontab = _aliases.get(crontab, crontab)
matchers = [_Matcher(which, entry)
for which, entry in enumerate(crontab.split())]
if len(matchers) == 5:
matchers.append(_Matcher(5, '*'))
_assert(len(matchers) == 6,
"improper number of cron entries specified")
matchers = Matcher(*matchers)
if not matchers.day.any:
_assert(matchers.weekday.any,
"missing a wildcard specifier for weekday")
if not matchers.weekday.any:
_assert(matchers.day.any,
"missing a wildcard specifier for day")
return matchers
def _test_match(self, index, dt):
'''
This tests the given field for whether it matches with the current
datetime object passed.
'''
at = _attribute[index]
attr = getattr(dt, at)
if at == 'isoweekday':
attr = attr() % 7
return self.matchers[index](attr, dt)
def next(self, now=None):
'''
How long to wait in seconds before this crontab entry can next be
executed.
'''
now = now or datetime.datetime.now()
# get a reasonable future start time
future = now.replace(second=0, microsecond=0) + datetime.timedelta(minutes=1)
to_test = 0
while to_test < 6:
incr = _increments[to_test]
while not self._test_match(to_test, future):
future += incr(future, self.matchers)
if self.matchers.year < future.year:
return None
# check for backtrack conditions
if to_test >= 3:
for tt in xrange(2, to_test):
if not self._test_match(tt, future):
# rely on the increment below to get us back to 2
to_test = 1
break
to_test += 1
# verify the match
match = [self._test_match(i, future) for i in xrange(6)]
_assert(all(match),
"\nYou have discovered a bug with crontab, please notify the\n" \
"author with the following information:\n" \
"crontab: %r\n" \
"now: %r", ' '.join(m.input for m in self.matchers), now)
delay = future - now
return delay.days * 86400 + delay.seconds + delay.microseconds / 1000000.

24
setup.py Normal file
View file

@ -0,0 +1,24 @@
#!/usr/bin/env python
from distutils.core import setup
with open('README') as f:
long_description = f.read()
setup(
name='crontab',
version='.1',
description='Parse and use crontab schedules in Python',
author='Josiah Carlson',
author_email='josiah.carlson@gmail.com',
url='https://github.com/josiahcarlson/parse-crontab',
packages=['crontab', 'tests'],
classifiers=[
'Development Status :: 4 - Beta',
'License :: OSI Approved :: GNU General Public License (GPL)',
'Programming Language :: Python',
],
license='GNU GPL v2.0',
long_description=long_description,
data_files=['Makefile'],
)

0
tests/__init__.py Normal file
View file

105
tests/test_crontab.py Normal file
View file

@ -0,0 +1,105 @@
import datetime
import unittest
from crontab import CronTab
class TestCrontab(unittest.TestCase):
def _run_test(self, crontab, max_delay, now=None):
ct = CronTab(crontab)
now = now or datetime.datetime.now()
delay = ct.next(now)
assert delay is not None
dd = (crontab, delay, max_delay, now, now+datetime.timedelta(seconds=delay))
assert delay <= max_delay, dd
def _run_impossible(self, crontab, now):
ct = CronTab(crontab)
delay = ct.next(now)
assert delay is None, (crontab, delay, now, now+datetime.timedelta(seconds=delay))
def test_normal(self):
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 * * * *', 60, 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))
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))
def test_last_day(self):
self._run_test('0 0 L 2 ?', 28*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))
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 * * * *'))
def test():
suite = unittest.TestLoader().loadTestsFromTestCase(TestCrontab)
unittest.TextTestRunner(verbosity=2).run(suite)
if __name__ == '__main__':
test()