fields.py 3.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100
  1. """Fields represent CronTrigger options which map to :class:`~datetime.datetime` fields."""
  2. from calendar import monthrange
  3. from apscheduler.triggers.cron.expressions import (
  4. AllExpression, RangeExpression, WeekdayPositionExpression, LastDayOfMonthExpression,
  5. WeekdayRangeExpression)
  6. __all__ = ('MIN_VALUES', 'MAX_VALUES', 'DEFAULT_VALUES', 'BaseField', 'WeekField',
  7. 'DayOfMonthField', 'DayOfWeekField')
  8. MIN_VALUES = {'year': 1970, 'month': 1, 'day': 1, 'week': 1, 'day_of_week': 0, 'hour': 0,
  9. 'minute': 0, 'second': 0}
  10. MAX_VALUES = {'year': 2 ** 63, 'month': 12, 'day:': 31, 'week': 53, 'day_of_week': 6, 'hour': 23,
  11. 'minute': 59, 'second': 59}
  12. DEFAULT_VALUES = {'year': '*', 'month': 1, 'day': 1, 'week': '*', 'day_of_week': '*', 'hour': 0,
  13. 'minute': 0, 'second': 0}
  14. class BaseField(object):
  15. REAL = True
  16. COMPILERS = [AllExpression, RangeExpression]
  17. def __init__(self, name, exprs, is_default=False):
  18. self.name = name
  19. self.is_default = is_default
  20. self.compile_expressions(exprs)
  21. def get_min(self, dateval):
  22. return MIN_VALUES[self.name]
  23. def get_max(self, dateval):
  24. return MAX_VALUES[self.name]
  25. def get_value(self, dateval):
  26. return getattr(dateval, self.name)
  27. def get_next_value(self, dateval):
  28. smallest = None
  29. for expr in self.expressions:
  30. value = expr.get_next_value(dateval, self)
  31. if smallest is None or (value is not None and value < smallest):
  32. smallest = value
  33. return smallest
  34. def compile_expressions(self, exprs):
  35. self.expressions = []
  36. # Split a comma-separated expression list, if any
  37. exprs = str(exprs).strip()
  38. if ',' in exprs:
  39. for expr in exprs.split(','):
  40. self.compile_expression(expr)
  41. else:
  42. self.compile_expression(exprs)
  43. def compile_expression(self, expr):
  44. for compiler in self.COMPILERS:
  45. match = compiler.value_re.match(expr)
  46. if match:
  47. compiled_expr = compiler(**match.groupdict())
  48. self.expressions.append(compiled_expr)
  49. return
  50. raise ValueError('Unrecognized expression "%s" for field "%s"' % (expr, self.name))
  51. def __eq__(self, other):
  52. return isinstance(self, self.__class__) and self.expressions == other.expressions
  53. def __str__(self):
  54. expr_strings = (str(e) for e in self.expressions)
  55. return ','.join(expr_strings)
  56. def __repr__(self):
  57. return "%s('%s', '%s')" % (self.__class__.__name__, self.name, self)
  58. class WeekField(BaseField):
  59. REAL = False
  60. def get_value(self, dateval):
  61. return dateval.isocalendar()[1]
  62. class DayOfMonthField(BaseField):
  63. COMPILERS = BaseField.COMPILERS + [WeekdayPositionExpression, LastDayOfMonthExpression]
  64. def get_max(self, dateval):
  65. return monthrange(dateval.year, dateval.month)[1]
  66. class DayOfWeekField(BaseField):
  67. REAL = False
  68. COMPILERS = BaseField.COMPILERS + [WeekdayRangeExpression]
  69. def get_value(self, dateval):
  70. return dateval.weekday()