diff --git a/.gitignore b/.gitignore index 0d20b64..9f5ab4e 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,4 @@ *.pyc +MANIFEST +build/* +dist/* diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..5ca2e71 --- /dev/null +++ b/Makefile @@ -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 diff --git a/README b/README new file mode 100644 index 0000000..1ba02ee --- /dev/null +++ b/README @@ -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 diff --git a/crontab/__init__.py b/crontab/__init__.py new file mode 100644 index 0000000..dee627b --- /dev/null +++ b/crontab/__init__.py @@ -0,0 +1,4 @@ + +from ._crontab import CronTab + +__all__ = ['CronTab'] diff --git a/crontab/_crontab.py b/crontab/_crontab.py new file mode 100644 index 0000000..17e03cc --- /dev/null +++ b/crontab/_crontab.py @@ -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. diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..3a9c434 --- /dev/null +++ b/setup.py @@ -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'], +) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_crontab.py b/tests/test_crontab.py new file mode 100644 index 0000000..c6c2abe --- /dev/null +++ b/tests/test_crontab.py @@ -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()