diff options
author | kiorky <kiorky@cryptelium.net> | 2017-08-31 09:36:57 +0200 |
---|---|---|
committer | GitHub <noreply@github.com> | 2017-08-31 09:36:57 +0200 |
commit | e46f26bdb9522193cdfbc7286df51669eb83ec82 (patch) | |
tree | 982cda8def08aa3bbf3f5f68f35b8821de33501d | |
parent | 1f4ad0b7fb0f0886db245091c4d627a4218d966f (diff) | |
parent | 5f98fb2fea1c3b791fcfa29a72ee9b13ea68c58d (diff) | |
download | croniter-e46f26bdb9522193cdfbc7286df51669eb83ec82.tar.gz |
Merge pull request #18 from taichino/master
back
-rw-r--r-- | README.rst | 5 | ||||
-rw-r--r-- | src/croniter/croniter.py | 226 | ||||
-rwxr-xr-x | src/croniter/tests/test_croniter.py | 16 |
3 files changed, 142 insertions, 105 deletions
@@ -82,6 +82,11 @@ Supported added for ``get_prev`` method. (>= 0.2.0):: >>> print itr.get_prev(datetime) # 2010-07-01 00:00:00 >>> print itr.get_prev(datetime) # 2010-06-01 00:00:00 +You can validate your crons using ``is_valid`` class method. (>= 0.3.18):: + + >>> croniter.is_valid('0 0 1 * *') # True + >>> croniter.is_valid('0 wrong_value 1 * *') # False + About DST ========= Be sure to init your croniter instance with a TZ aware datetime for this to work !:: diff --git a/src/croniter/croniter.py b/src/croniter/croniter.py index 55bce86..a9f10a3 100644 --- a/src/croniter/croniter.py +++ b/src/croniter/croniter.py @@ -14,18 +14,23 @@ search_re = re.compile(r'^([^-]+)-([^-/]+)(/(.*))?$') only_int_re = re.compile(r'^\d+$') any_int_re = re.compile(r'^\d+') star_or_int_re = re.compile(r'^(\d+|\*)$') +VALID_LEN_EXPRESSION = [5, 6] -class CroniterBadCronError(ValueError): - '''.''' +class CroniterError(ValueError): + pass -class CroniterBadDateError(ValueError): - '''.''' +class CroniterBadCronError(CroniterError): + pass -class CroniterNotAlphaError(ValueError): - '''.''' +class CroniterBadDateError(CroniterError): + pass + + +class CroniterNotAlphaError(CroniterError): + pass class croniter(object): @@ -79,106 +84,16 @@ class croniter(object): self.start_time = start_time self.cur = start_time - self.exprs = expr_format.split() - - if len(self.exprs) != 5 and len(self.exprs) != 6: - raise CroniterBadCronError(self.bad_length) - - expanded = [] - nth_weekday_of_month = {} - - for i, expr in enumerate(self.exprs): - e_list = expr.split(',') - res = [] - - while len(e_list) > 0: - e = e_list.pop() - - if i == 4: - e, sep, nth = str(e).partition('#') - if nth and not re.match(r'[1-5]', nth): - raise CroniterBadDateError( - "[{0}] is not acceptable".format(expr_format)) - - t = re.sub(r'^\*(\/.+)$', r'%d-%d\1' % ( - self.RANGES[i][0], - self.RANGES[i][1]), - str(e)) - m = search_re.search(t) - if not m: - t = re.sub(r'^(.+)\/(.+)$', r'\1-%d/\2' % ( - self.RANGES[i][1]), - str(e)) - m = step_search_re.search(t) + self.expanded, self.nth_weekday_of_month = self.expand(expr_format) - if m: - (low, high, step) = m.group(1), m.group(2), m.group(4) or 1 - - if not any_int_re.search(low): - low = "{0}".format(self._alphaconv(i, low)) - - if not any_int_re.search(high): - high = "{0}".format(self._alphaconv(i, high)) - - if ( - not low or not high or int(low) > int(high) - or not only_int_re.search(str(step)) - ): - raise CroniterBadDateError( - "[{0}] is not acceptable".format(expr_format)) - - low, high, step = map(int, [low, high, step]) - rng = range(low, high + 1, step) - e_list += (["{0}#{1}".format(item, nth) for item in rng] - if i == 4 and nth else rng) - else: - if t.startswith('-'): - raise CroniterBadCronError( - "[{0}] is not acceptable,\ - negative numbers not allowed".format( - expr_format)) - if not star_or_int_re.search(t): - t = self._alphaconv(i, t) - - try: - t = int(t) - except: - pass - - if t in self.LOWMAP[i]: - t = self.LOWMAP[i][t] - - if ( - t not in ["*", "l"] - and (int(t) < self.RANGES[i][0] or - int(t) > self.RANGES[i][1]) - ): - raise CroniterBadCronError( - "[{0}] is not acceptable, out of range".format( - expr_format)) - - res.append(t) - - if i == 4 and nth: - if t not in nth_weekday_of_month: - nth_weekday_of_month[t] = set() - nth_weekday_of_month[t].add(int(nth)) - - res.sort() - expanded.append(['*'] if (len(res) == 1 - and res[0] == '*') - else res) - - self.expanded = expanded - self.nth_weekday_of_month = nth_weekday_of_month - - def _alphaconv(self, index, key): + @classmethod + def _alphaconv(cls, index, key, expressions): try: - return self.ALPHACONV[index][key.lower()] + return cls.ALPHACONV[index][key.lower()] except KeyError: raise CroniterNotAlphaError( - "[{0}] is not acceptable".format(" ".join(self.exprs))) + "[{0}] is not acceptable".format(" ".join(expressions))) def get_next(self, ret_type=None): return self._get_next(ret_type or self._ret_type, is_prev=False) @@ -192,14 +107,15 @@ class croniter(object): return self._timestamp_to_datetime(self.cur) return self.cur - def _datetime_to_timestamp(self, d): + @classmethod + def _datetime_to_timestamp(cls, d): """ Converts a `datetime` object `d` into a UNIX timestamp. """ if d.tzinfo is not None: d = d.replace(tzinfo=None) - d.utcoffset() - return self._timedelta_to_seconds(d - datetime.datetime(1970, 1, 1)) + return cls._timedelta_to_seconds(d - datetime.datetime(1970, 1, 1)) def _timestamp_to_datetime(self, timestamp): """ @@ -532,3 +448,107 @@ class croniter(object): return True else: return False + + @classmethod + def expand(cls, expr_format): + expressions = expr_format.split() + + if len(expressions) not in VALID_LEN_EXPRESSION: + raise CroniterBadCronError(cls.bad_length) + + expanded = [] + nth_weekday_of_month = {} + + for i, expr in enumerate(expressions): + e_list = expr.split(',') + res = [] + + while len(e_list) > 0: + e = e_list.pop() + + if i == 4: + e, sep, nth = str(e).partition('#') + if nth and not re.match(r'[1-5]', nth): + raise CroniterBadDateError( + "[{0}] is not acceptable".format(expr_format)) + + t = re.sub(r'^\*(\/.+)$', r'%d-%d\1' % ( + cls.RANGES[i][0], + cls.RANGES[i][1]), + str(e)) + m = search_re.search(t) + + if not m: + t = re.sub(r'^(.+)\/(.+)$', r'\1-%d/\2' % ( + cls.RANGES[i][1]), + str(e)) + m = step_search_re.search(t) + + if m: + (low, high, step) = m.group(1), m.group(2), m.group(4) or 1 + + if not any_int_re.search(low): + low = "{0}".format(cls._alphaconv(i, low, expressions)) + + if not any_int_re.search(high): + high = "{0}".format(cls._alphaconv(i, high, expressions)) + + if ( + not low or not high or int(low) > int(high) + or not only_int_re.search(str(step)) + ): + raise CroniterBadDateError( + "[{0}] is not acceptable".format(expr_format)) + + low, high, step = map(int, [low, high, step]) + rng = range(low, high + 1, step) + e_list += (["{0}#{1}".format(item, nth) for item in rng] + if i == 4 and nth else rng) + else: + if t.startswith('-'): + raise CroniterBadCronError( + "[{0}] is not acceptable,\ + negative numbers not allowed".format( + expr_format)) + if not star_or_int_re.search(t): + t = cls._alphaconv(i, t, expressions) + + try: + t = int(t) + except: + pass + + if t in cls.LOWMAP[i]: + t = cls.LOWMAP[i][t] + + if ( + t not in ["*", "l"] + and (int(t) < cls.RANGES[i][0] or + int(t) > cls.RANGES[i][1]) + ): + raise CroniterBadCronError( + "[{0}] is not acceptable, out of range".format( + expr_format)) + + res.append(t) + + if i == 4 and nth: + if t not in nth_weekday_of_month: + nth_weekday_of_month[t] = set() + nth_weekday_of_month[t].add(int(nth)) + + res.sort() + expanded.append(['*'] if (len(res) == 1 + and res[0] == '*') + else res) + + return expanded, nth_weekday_of_month + + @classmethod + def is_valid(cls, expression): + try: + cls.expand(expression) + except CroniterError: + return False + else: + return True diff --git a/src/croniter/tests/test_croniter.py b/src/croniter/tests/test_croniter.py index 9238daa..267791d 100755 --- a/src/croniter/tests/test_croniter.py +++ b/src/croniter/tests/test_croniter.py @@ -2,10 +2,10 @@ # -*- coding: utf-8 -*- import unittest -from datetime import datetime, timedelta +from datetime import datetime from time import sleep import pytz -from croniter import croniter, CroniterBadDateError +from croniter import croniter, CroniterBadDateError, CroniterBadCronError, CroniterNotAlphaError from croniter.tests import base @@ -733,5 +733,17 @@ class CroniterTest(base.TestCase): val = croniter('0 * * * *', local_date).get_next(datetime) self.assertEqual(val, tz.localize(datetime(2017, 10, 29, 6))) + def test_error_alpha_cron(self): + self.assertRaises(CroniterNotAlphaError, croniter.expand, '* * * janu-jun *') + + def test_error_bad_cron(self): + self.assertRaises(CroniterBadCronError, croniter.expand, '* * * *') + self.assertRaises(CroniterBadCronError, croniter.expand, '* * * * * * *') + + def test_is_valid(self): + self.assertTrue(croniter.is_valid('0 * * * *')) + self.assertFalse(croniter.is_valid('0 * *')) + self.assertFalse(croniter.is_valid('* * * janu-jun *')) + if __name__ == '__main__': unittest.main() |