diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/fields.py')
-rw-r--r-- | .venv/lib/python3.12/site-packages/apscheduler/triggers/cron/fields.py | 149 |
1 files changed, 149 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/fields.py b/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/fields.py new file mode 100644 index 00000000..4c081797 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/fields.py @@ -0,0 +1,149 @@ +"""Fields represent CronTrigger options which map to :class:`~datetime.datetime` fields.""" + +__all__ = ( + "MIN_VALUES", + "MAX_VALUES", + "DEFAULT_VALUES", + "BaseField", + "WeekField", + "DayOfMonthField", + "DayOfWeekField", +) + +import re +from calendar import monthrange + +from apscheduler.triggers.cron.expressions import ( + AllExpression, + LastDayOfMonthExpression, + MonthRangeExpression, + RangeExpression, + WeekdayPositionExpression, + WeekdayRangeExpression, +) + +MIN_VALUES = { + "year": 1970, + "month": 1, + "day": 1, + "week": 1, + "day_of_week": 0, + "hour": 0, + "minute": 0, + "second": 0, +} +MAX_VALUES = { + "year": 9999, + "month": 12, + "day": 31, + "week": 53, + "day_of_week": 6, + "hour": 23, + "minute": 59, + "second": 59, +} +DEFAULT_VALUES = { + "year": "*", + "month": 1, + "day": 1, + "week": "*", + "day_of_week": "*", + "hour": 0, + "minute": 0, + "second": 0, +} +SEPARATOR = re.compile(" *, *") + + +class BaseField: + REAL = True + COMPILERS = [AllExpression, RangeExpression] + + def __init__(self, name, exprs, is_default=False): + self.name = name + self.is_default = is_default + self.compile_expressions(exprs) + + def get_min(self, dateval): + return MIN_VALUES[self.name] + + def get_max(self, dateval): + return MAX_VALUES[self.name] + + def get_value(self, dateval): + return getattr(dateval, self.name) + + def get_next_value(self, dateval): + smallest = None + for expr in self.expressions: + value = expr.get_next_value(dateval, self) + if smallest is None or (value is not None and value < smallest): + smallest = value + + return smallest + + def compile_expressions(self, exprs): + self.expressions = [] + + # Split a comma-separated expression list, if any + for expr in SEPARATOR.split(str(exprs).strip()): + self.compile_expression(expr) + + def compile_expression(self, expr): + for compiler in self.COMPILERS: + match = compiler.value_re.match(expr) + if match: + compiled_expr = compiler(**match.groupdict()) + + try: + compiled_expr.validate_range(self.name) + except ValueError as e: + raise ValueError( + f"Error validating expression {expr!r}: {e}" + ) from None + + self.expressions.append(compiled_expr) + return + + raise ValueError(f'Unrecognized expression "{expr}" for field "{self.name}"') + + def __eq__(self, other): + return ( + isinstance(self, self.__class__) and self.expressions == other.expressions + ) + + def __str__(self): + expr_strings = (str(e) for e in self.expressions) + return ",".join(expr_strings) + + def __repr__(self): + return f"{self.__class__.__name__}('{self.name}', '{self}')" + + +class WeekField(BaseField): + REAL = False + + def get_value(self, dateval): + return dateval.isocalendar()[1] + + +class DayOfMonthField(BaseField): + COMPILERS = BaseField.COMPILERS + [ + WeekdayPositionExpression, + LastDayOfMonthExpression, + ] + + def get_max(self, dateval): + return monthrange(dateval.year, dateval.month)[1] + + +class DayOfWeekField(BaseField): + REAL = False + COMPILERS = BaseField.COMPILERS + [WeekdayRangeExpression] + + def get_value(self, dateval): + return dateval.weekday() + + +class MonthField(BaseField): + COMPILERS = BaseField.COMPILERS + [MonthRangeExpression] |