diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/apscheduler/triggers/cron')
3 files changed, 723 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/__init__.py b/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/__init__.py new file mode 100644 index 00000000..03be8196 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/__init__.py @@ -0,0 +1,289 @@ +from datetime import datetime, timedelta + +from tzlocal import get_localzone + +from apscheduler.triggers.base import BaseTrigger +from apscheduler.triggers.cron.fields import ( + DEFAULT_VALUES, + BaseField, + DayOfMonthField, + DayOfWeekField, + MonthField, + WeekField, +) +from apscheduler.util import ( + astimezone, + convert_to_datetime, + datetime_ceil, + datetime_repr, +) + + +class CronTrigger(BaseTrigger): + """ + Triggers when current time matches all specified time constraints, + similarly to how the UNIX cron scheduler works. + + :param int|str year: 4-digit year + :param int|str month: month (1-12) + :param int|str day: day of month (1-31) + :param int|str week: ISO week (1-53) + :param int|str day_of_week: number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun) + :param int|str hour: hour (0-23) + :param int|str minute: minute (0-59) + :param int|str second: second (0-59) + :param datetime|str start_date: earliest possible date/time to trigger on (inclusive) + :param datetime|str end_date: latest possible date/time to trigger on (inclusive) + :param datetime.tzinfo|str timezone: time zone to use for the date/time calculations (defaults + to scheduler timezone) + :param int|None jitter: delay the job execution by ``jitter`` seconds at most + + .. note:: The first weekday is always **monday**. + """ + + FIELD_NAMES = ( + "year", + "month", + "day", + "week", + "day_of_week", + "hour", + "minute", + "second", + ) + FIELDS_MAP = { + "year": BaseField, + "month": MonthField, + "week": WeekField, + "day": DayOfMonthField, + "day_of_week": DayOfWeekField, + "hour": BaseField, + "minute": BaseField, + "second": BaseField, + } + + __slots__ = "timezone", "start_date", "end_date", "fields", "jitter" + + def __init__( + self, + year=None, + month=None, + day=None, + week=None, + day_of_week=None, + hour=None, + minute=None, + second=None, + start_date=None, + end_date=None, + timezone=None, + jitter=None, + ): + if timezone: + self.timezone = astimezone(timezone) + elif isinstance(start_date, datetime) and start_date.tzinfo: + self.timezone = astimezone(start_date.tzinfo) + elif isinstance(end_date, datetime) and end_date.tzinfo: + self.timezone = astimezone(end_date.tzinfo) + else: + self.timezone = get_localzone() + + self.start_date = convert_to_datetime(start_date, self.timezone, "start_date") + self.end_date = convert_to_datetime(end_date, self.timezone, "end_date") + + self.jitter = jitter + + values = dict( + (key, value) + for (key, value) in locals().items() + if key in self.FIELD_NAMES and value is not None + ) + self.fields = [] + assign_defaults = False + for field_name in self.FIELD_NAMES: + if field_name in values: + exprs = values.pop(field_name) + is_default = False + assign_defaults = not values + elif assign_defaults: + exprs = DEFAULT_VALUES[field_name] + is_default = True + else: + exprs = "*" + is_default = True + + field_class = self.FIELDS_MAP[field_name] + field = field_class(field_name, exprs, is_default) + self.fields.append(field) + + @classmethod + def from_crontab(cls, expr, timezone=None): + """ + Create a :class:`~CronTrigger` from a standard crontab expression. + + See https://en.wikipedia.org/wiki/Cron for more information on the format accepted here. + + :param expr: minute, hour, day of month, month, day of week + :param datetime.tzinfo|str timezone: time zone to use for the date/time calculations ( + defaults to scheduler timezone) + :return: a :class:`~CronTrigger` instance + + """ + values = expr.split() + if len(values) != 5: + raise ValueError(f"Wrong number of fields; got {len(values)}, expected 5") + + return cls( + minute=values[0], + hour=values[1], + day=values[2], + month=values[3], + day_of_week=values[4], + timezone=timezone, + ) + + def _increment_field_value(self, dateval, fieldnum): + """ + Increments the designated field and resets all less significant fields to their minimum + values. + + :type dateval: datetime + :type fieldnum: int + :return: a tuple containing the new date, and the number of the field that was actually + incremented + :rtype: tuple + """ + + values = {} + i = 0 + while i < len(self.fields): + field = self.fields[i] + if not field.REAL: + if i == fieldnum: + fieldnum -= 1 + i -= 1 + else: + i += 1 + continue + + if i < fieldnum: + values[field.name] = field.get_value(dateval) + i += 1 + elif i > fieldnum: + values[field.name] = field.get_min(dateval) + i += 1 + else: + value = field.get_value(dateval) + maxval = field.get_max(dateval) + if value == maxval: + fieldnum -= 1 + i -= 1 + else: + values[field.name] = value + 1 + i += 1 + + difference = datetime(**values) - dateval.replace(tzinfo=None) + dateval = datetime.fromtimestamp( + dateval.timestamp() + difference.total_seconds(), self.timezone + ) + return dateval, fieldnum + + def _set_field_value(self, dateval, fieldnum, new_value): + values = {} + for i, field in enumerate(self.fields): + if field.REAL: + if i < fieldnum: + values[field.name] = field.get_value(dateval) + elif i > fieldnum: + values[field.name] = field.get_min(dateval) + else: + values[field.name] = new_value + + return datetime(**values, tzinfo=self.timezone, fold=dateval.fold) + + def get_next_fire_time(self, previous_fire_time, now): + if previous_fire_time: + start_date = min(now, previous_fire_time + timedelta(microseconds=1)) + if start_date == previous_fire_time: + start_date += timedelta(microseconds=1) + else: + start_date = max(now, self.start_date) if self.start_date else now + + fieldnum = 0 + next_date = datetime_ceil(start_date).astimezone(self.timezone) + while 0 <= fieldnum < len(self.fields): + field = self.fields[fieldnum] + curr_value = field.get_value(next_date) + next_value = field.get_next_value(next_date) + + if next_value is None: + # No valid value was found + next_date, fieldnum = self._increment_field_value( + next_date, fieldnum - 1 + ) + elif next_value > curr_value: + # A valid, but higher than the starting value, was found + if field.REAL: + next_date = self._set_field_value(next_date, fieldnum, next_value) + fieldnum += 1 + else: + next_date, fieldnum = self._increment_field_value( + next_date, fieldnum + ) + else: + # A valid value was found, no changes necessary + fieldnum += 1 + + # Return if the date has rolled past the end date + if self.end_date and next_date > self.end_date: + return None + + if fieldnum >= 0: + next_date = self._apply_jitter(next_date, self.jitter, now) + return min(next_date, self.end_date) if self.end_date else next_date + + def __getstate__(self): + return { + "version": 2, + "timezone": self.timezone, + "start_date": self.start_date, + "end_date": self.end_date, + "fields": self.fields, + "jitter": self.jitter, + } + + def __setstate__(self, state): + # This is for compatibility with APScheduler 3.0.x + if isinstance(state, tuple): + state = state[1] + + if state.get("version", 1) > 2: + raise ValueError( + f"Got serialized data for version {state['version']} of " + f"{self.__class__.__name__}, but only versions up to 2 can be handled" + ) + + self.timezone = astimezone(state["timezone"]) + self.start_date = state["start_date"] + self.end_date = state["end_date"] + self.fields = state["fields"] + self.jitter = state.get("jitter") + + def __str__(self): + options = [f"{f.name}='{f}'" for f in self.fields if not f.is_default] + return "cron[{}]".format(", ".join(options)) + + def __repr__(self): + options = [f"{f.name}='{f}'" for f in self.fields if not f.is_default] + if self.start_date: + options.append(f"start_date={datetime_repr(self.start_date)!r}") + if self.end_date: + options.append(f"end_date={datetime_repr(self.end_date)!r}") + if self.jitter: + options.append(f"jitter={self.jitter}") + + return "<{} ({}, timezone='{}')>".format( + self.__class__.__name__, + ", ".join(options), + self.timezone, + ) diff --git a/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/expressions.py b/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/expressions.py new file mode 100644 index 00000000..0d84ec23 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/apscheduler/triggers/cron/expressions.py @@ -0,0 +1,285 @@ +"""This module contains the expressions applicable for CronTrigger's fields.""" + +__all__ = ( + "AllExpression", + "RangeExpression", + "WeekdayRangeExpression", + "WeekdayPositionExpression", + "LastDayOfMonthExpression", +) + +import re +from calendar import monthrange + +from apscheduler.util import asint + +WEEKDAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"] +MONTHS = [ + "jan", + "feb", + "mar", + "apr", + "may", + "jun", + "jul", + "aug", + "sep", + "oct", + "nov", + "dec", +] + + +class AllExpression: + value_re = re.compile(r"\*(?:/(?P<step>\d+))?$") + + def __init__(self, step=None): + self.step = asint(step) + if self.step == 0: + raise ValueError("Increment must be higher than 0") + + def validate_range(self, field_name): + from apscheduler.triggers.cron.fields import MAX_VALUES, MIN_VALUES + + value_range = MAX_VALUES[field_name] - MIN_VALUES[field_name] + if self.step and self.step > value_range: + raise ValueError( + f"the step value ({self.step}) is higher than the total range of the " + f"expression ({value_range})" + ) + + def get_next_value(self, date, field): + start = field.get_value(date) + minval = field.get_min(date) + maxval = field.get_max(date) + start = max(start, minval) + + if not self.step: + next = start + else: + distance_to_next = (self.step - (start - minval)) % self.step + next = start + distance_to_next + + if next <= maxval: + return next + + def __eq__(self, other): + return isinstance(other, self.__class__) and self.step == other.step + + def __str__(self): + if self.step: + return "*/%d" % self.step + return "*" + + def __repr__(self): + return f"{self.__class__.__name__}({self.step})" + + +class RangeExpression(AllExpression): + value_re = re.compile(r"(?P<first>\d+)(?:-(?P<last>\d+))?(?:/(?P<step>\d+))?$") + + def __init__(self, first, last=None, step=None): + super().__init__(step) + first = asint(first) + last = asint(last) + if last is None and step is None: + last = first + if last is not None and first > last: + raise ValueError( + "The minimum value in a range must not be higher than the maximum" + ) + self.first = first + self.last = last + + def validate_range(self, field_name): + from apscheduler.triggers.cron.fields import MAX_VALUES, MIN_VALUES + + super().validate_range(field_name) + if self.first < MIN_VALUES[field_name]: + raise ValueError( + f"the first value ({self.first}) is lower than the minimum value ({MIN_VALUES[field_name]})" + ) + if self.last is not None and self.last > MAX_VALUES[field_name]: + raise ValueError( + f"the last value ({self.last}) is higher than the maximum value ({MAX_VALUES[field_name]})" + ) + value_range = (self.last or MAX_VALUES[field_name]) - self.first + if self.step and self.step > value_range: + raise ValueError( + f"the step value ({self.step}) is higher than the total range of the " + f"expression ({value_range})" + ) + + def get_next_value(self, date, field): + startval = field.get_value(date) + minval = field.get_min(date) + maxval = field.get_max(date) + + # Apply range limits + minval = max(minval, self.first) + maxval = min(maxval, self.last) if self.last is not None else maxval + nextval = max(minval, startval) + + # Apply the step if defined + if self.step: + distance_to_next = (self.step - (nextval - minval)) % self.step + nextval += distance_to_next + + return nextval if nextval <= maxval else None + + def __eq__(self, other): + return ( + isinstance(other, self.__class__) + and self.first == other.first + and self.last == other.last + ) + + def __str__(self): + if self.last != self.first and self.last is not None: + range = "%d-%d" % (self.first, self.last) + else: + range = str(self.first) + + if self.step: + return "%s/%d" % (range, self.step) + + return range + + def __repr__(self): + args = [str(self.first)] + if self.last != self.first and self.last is not None or self.step: + args.append(str(self.last)) + + if self.step: + args.append(str(self.step)) + + return "{}({})".format(self.__class__.__name__, ", ".join(args)) + + +class MonthRangeExpression(RangeExpression): + value_re = re.compile(r"(?P<first>[a-z]+)(?:-(?P<last>[a-z]+))?", re.IGNORECASE) + + def __init__(self, first, last=None): + try: + first_num = MONTHS.index(first.lower()) + 1 + except ValueError: + raise ValueError(f'Invalid month name "{first}"') + + if last: + try: + last_num = MONTHS.index(last.lower()) + 1 + except ValueError: + raise ValueError(f'Invalid month name "{last}"') + else: + last_num = None + + super().__init__(first_num, last_num) + + def __str__(self): + if self.last != self.first and self.last is not None: + return f"{MONTHS[self.first - 1]}-{MONTHS[self.last - 1]}" + return MONTHS[self.first - 1] + + def __repr__(self): + args = [f"'{MONTHS[self.first]}'"] + if self.last != self.first and self.last is not None: + args.append(f"'{MONTHS[self.last - 1]}'") + return "{}({})".format(self.__class__.__name__, ", ".join(args)) + + +class WeekdayRangeExpression(RangeExpression): + value_re = re.compile(r"(?P<first>[a-z]+)(?:-(?P<last>[a-z]+))?", re.IGNORECASE) + + def __init__(self, first, last=None): + try: + first_num = WEEKDAYS.index(first.lower()) + except ValueError: + raise ValueError(f'Invalid weekday name "{first}"') + + if last: + try: + last_num = WEEKDAYS.index(last.lower()) + except ValueError: + raise ValueError(f'Invalid weekday name "{last}"') + else: + last_num = None + + super().__init__(first_num, last_num) + + def __str__(self): + if self.last != self.first and self.last is not None: + return f"{WEEKDAYS[self.first]}-{WEEKDAYS[self.last]}" + return WEEKDAYS[self.first] + + def __repr__(self): + args = [f"'{WEEKDAYS[self.first]}'"] + if self.last != self.first and self.last is not None: + args.append(f"'{WEEKDAYS[self.last]}'") + return "{}({})".format(self.__class__.__name__, ", ".join(args)) + + +class WeekdayPositionExpression(AllExpression): + options = ["1st", "2nd", "3rd", "4th", "5th", "last"] + value_re = re.compile( + r"(?P<option_name>{}) +(?P<weekday_name>(?:\d+|\w+))".format("|".join(options)), + re.IGNORECASE, + ) + + def __init__(self, option_name, weekday_name): + super().__init__(None) + try: + self.option_num = self.options.index(option_name.lower()) + except ValueError: + raise ValueError(f'Invalid weekday position "{option_name}"') + + try: + self.weekday = WEEKDAYS.index(weekday_name.lower()) + except ValueError: + raise ValueError(f'Invalid weekday name "{weekday_name}"') + + def get_next_value(self, date, field): + # Figure out the weekday of the month's first day and the number of days in that month + first_day_wday, last_day = monthrange(date.year, date.month) + + # Calculate which day of the month is the first of the target weekdays + first_hit_day = self.weekday - first_day_wday + 1 + if first_hit_day <= 0: + first_hit_day += 7 + + # Calculate what day of the month the target weekday would be + if self.option_num < 5: + target_day = first_hit_day + self.option_num * 7 + else: + target_day = first_hit_day + ((last_day - first_hit_day) // 7) * 7 + + if target_day <= last_day and target_day >= date.day: + return target_day + + def __eq__(self, other): + return ( + super().__eq__(other) + and self.option_num == other.option_num + and self.weekday == other.weekday + ) + + def __str__(self): + return f"{self.options[self.option_num]} {WEEKDAYS[self.weekday]}" + + def __repr__(self): + return f"{self.__class__.__name__}('{self.options[self.option_num]}', '{WEEKDAYS[self.weekday]}')" + + +class LastDayOfMonthExpression(AllExpression): + value_re = re.compile(r"last", re.IGNORECASE) + + def __init__(self): + super().__init__(None) + + def get_next_value(self, date, field): + return monthrange(date.year, date.month)[1] + + def __str__(self): + return "last" + + def __repr__(self): + return f"{self.__class__.__name__}()" 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] |