diff options
Diffstat (limited to 'src/apscheduler/triggers')
-rw-r--r-- | src/apscheduler/triggers/calendarinterval.py | 6 | ||||
-rw-r--r-- | src/apscheduler/triggers/combining.py | 20 | ||||
-rw-r--r-- | src/apscheduler/triggers/cron/__init__.py | 10 | ||||
-rw-r--r-- | src/apscheduler/triggers/cron/expressions.py | 7 | ||||
-rw-r--r-- | src/apscheduler/triggers/cron/fields.py | 6 | ||||
-rw-r--r-- | src/apscheduler/triggers/date.py | 4 | ||||
-rw-r--r-- | src/apscheduler/triggers/interval.py | 8 |
7 files changed, 30 insertions, 31 deletions
diff --git a/src/apscheduler/triggers/calendarinterval.py b/src/apscheduler/triggers/calendarinterval.py index 8d0990d..fbb5896 100644 --- a/src/apscheduler/triggers/calendarinterval.py +++ b/src/apscheduler/triggers/calendarinterval.py @@ -1,7 +1,7 @@ from __future__ import annotations from datetime import date, datetime, time, timedelta, tzinfo -from typing import Any, Optional +from typing import Any import attrs @@ -67,7 +67,7 @@ class CalendarIntervalTrigger(Trigger): end_date: date | None = attrs.field(converter=as_date, default=None) timezone: tzinfo = attrs.field(converter=as_timezone, default='local') _time: time = attrs.field(init=False, eq=False) - _last_fire_date: Optional[date] = attrs.field(init=False, eq=False, default=None) + _last_fire_date: date | None = attrs.field(init=False, eq=False, default=None) def __attrs_post_init__(self) -> None: self._time = time(self.hour, self.minute, self.second, tzinfo=self.timezone) @@ -78,7 +78,7 @@ class CalendarIntervalTrigger(Trigger): if self.start_date and self.end_date and self.start_date > self.end_date: raise ValueError('end_date cannot be earlier than start_date') - def next(self) -> Optional[datetime]: + def next(self) -> datetime | None: previous_date: date = self._last_fire_date while True: if previous_date: diff --git a/src/apscheduler/triggers/combining.py b/src/apscheduler/triggers/combining.py index fa0b5c9..bcee64a 100644 --- a/src/apscheduler/triggers/combining.py +++ b/src/apscheduler/triggers/combining.py @@ -2,7 +2,7 @@ from __future__ import annotations from abc import abstractmethod from datetime import datetime, timedelta -from typing import Any, Optional +from typing import Any import attrs @@ -15,7 +15,7 @@ from ..validators import as_timedelta, require_state_version @attrs.define class BaseCombiningTrigger(Trigger): triggers: list[Trigger] - _next_fire_times: list[Optional[datetime]] = attrs.field(init=False, eq=False, factory=list) + _next_fire_times: list[datetime | None] = attrs.field(init=False, eq=False, factory=list) def __getstate__(self) -> dict[str, Any]: return { @@ -51,17 +51,17 @@ class AndTrigger(BaseCombiningTrigger): """ threshold: timedelta = attrs.field(converter=as_timedelta, default=1) - max_iterations: Optional[int] = 10000 + max_iterations: int | None = 10000 - def next(self) -> Optional[datetime]: + def next(self) -> datetime | None: if not self._next_fire_times: # Fill out the fire times on the first run self._next_fire_times = [t.next() for t in self.triggers] for _ in range(self.max_iterations): # Find the earliest and latest fire times - earliest_fire_time: Optional[datetime] = None - latest_fire_time: Optional[datetime] = None + earliest_fire_time: datetime | None = None + latest_fire_time: datetime | None = None for fire_time in self._next_fire_times: # If any of the fire times is None, this trigger is finished if fire_time is None: @@ -74,7 +74,7 @@ class AndTrigger(BaseCombiningTrigger): latest_fire_time = fire_time # Replace all the fire times that were within the threshold - for i, trigger in enumerate(self.triggers): + for i, _trigger in enumerate(self.triggers): if self._next_fire_times[i] - earliest_fire_time <= self.threshold: self._next_fire_times[i] = self.triggers[i].next() @@ -114,14 +114,14 @@ class OrTrigger(BaseCombiningTrigger): :param triggers: triggers to combine """ - def next(self) -> Optional[datetime]: + def next(self) -> datetime | None: # Fill out the fire times on the first run if not self._next_fire_times: self._next_fire_times = [t.next() for t in self.triggers] # Find out the earliest of the fire times - earliest_time: Optional[datetime] = min([fire_time for fire_time in self._next_fire_times - if fire_time is not None], default=None) + earliest_time: datetime | None = min((fire_time for fire_time in self._next_fire_times + if fire_time is not None), default=None) if earliest_time is not None: # Generate new fire times for the trigger(s) that generated the earliest fire time for i, fire_time in enumerate(self._next_fire_times): diff --git a/src/apscheduler/triggers/cron/__init__.py b/src/apscheduler/triggers/cron/__init__.py index 1c1e582..df1f9e5 100644 --- a/src/apscheduler/triggers/cron/__init__.py +++ b/src/apscheduler/triggers/cron/__init__.py @@ -1,7 +1,7 @@ from __future__ import annotations from datetime import datetime, timedelta, tzinfo -from typing import Any, ClassVar, Optional, Sequence +from typing import Any, ClassVar, Sequence import attrs from tzlocal import get_localzone @@ -59,12 +59,12 @@ class CronTrigger(Trigger): end_time: datetime | None = None timezone: tzinfo | str = attrs.field(converter=as_timezone, factory=get_localzone) _fields: list[BaseField] = attrs.field(init=False, eq=False, factory=list) - _last_fire_time: Optional[datetime] = attrs.field(init=False, eq=False, default=None) + _last_fire_time: datetime | None = attrs.field(init=False, eq=False, default=None) def __attrs_post_init__(self) -> None: self._set_fields([self.year, self.month, self.day, self.week, self.day_of_week, self.hour, self.minute, self.second]) - self._last_fire_time: Optional[datetime] = None + self._last_fire_time: datetime | None = None def _set_fields(self, values: Sequence[int | str | None]) -> None: self._fields = [] @@ -80,7 +80,7 @@ class CronTrigger(Trigger): self._fields.append(field) @classmethod - def from_crontab(cls, expr: str, timezone: str | tzinfo = 'local') -> 'CronTrigger': + def from_crontab(cls, expr: str, timezone: str | tzinfo = 'local') -> CronTrigger: """ Create a :class:`~CronTrigger` from a standard crontab expression. @@ -153,7 +153,7 @@ class CronTrigger(Trigger): return datetime(**values, tzinfo=self.timezone) - def next(self) -> Optional[datetime]: + def next(self) -> datetime | None: if self._last_fire_time: start_time = self._last_fire_time + timedelta(microseconds=1) else: diff --git a/src/apscheduler/triggers/cron/expressions.py b/src/apscheduler/triggers/cron/expressions.py index e417dea..5c94afa 100644 --- a/src/apscheduler/triggers/cron/expressions.py +++ b/src/apscheduler/triggers/cron/expressions.py @@ -4,7 +4,6 @@ from __future__ import annotations import re from calendar import monthrange from datetime import datetime -from typing import Optional from ...validators import as_int @@ -35,7 +34,7 @@ class AllExpression: 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, dateval: datetime, field) -> Optional[int]: + def get_next_value(self, dateval: datetime, field) -> int | None: start = field.get_value(dateval) minval = field.get_min(dateval) maxval = field.get_max(dateval) @@ -144,7 +143,7 @@ class WeekdayRangeExpression(RangeExpression): value_re = re.compile(r'(?P<first>[a-z]+)(?:-(?P<last>[a-z]+))?', re.IGNORECASE) - def __init__(self, first: str, last: Optional[str] = None): + def __init__(self, first: str, last: str | None = None): first_num = get_weekday_index(first) last_num = get_weekday_index(last) if last else None super().__init__(first_num, last_num) @@ -171,7 +170,7 @@ class WeekdayPositionExpression(AllExpression): except ValueError: raise ValueError(f'Invalid weekday name {weekday_name!r}') from None - def get_next_value(self, dateval: datetime, field) -> Optional[int]: + def get_next_value(self, dateval: datetime, field) -> int | None: # Figure out the weekday of the month's first day and the number of days in that month first_day_wday, last_day = monthrange(dateval.year, dateval.month) diff --git a/src/apscheduler/triggers/cron/fields.py b/src/apscheduler/triggers/cron/fields.py index d15fcd5..e68fdca 100644 --- a/src/apscheduler/triggers/cron/fields.py +++ b/src/apscheduler/triggers/cron/fields.py @@ -4,7 +4,7 @@ from __future__ import annotations import re from calendar import monthrange from datetime import datetime -from typing import Any, ClassVar, List, Optional, Sequence +from typing import Any, ClassVar, Sequence from .expressions import ( WEEKDAYS, AllExpression, LastDayOfMonthExpression, MonthRangeExpression, RangeExpression, @@ -32,7 +32,7 @@ class BaseField: def __init__(self, name: str, exprs: int | str): self.name = name - self.expressions: List = [] + self.expressions: list = [] for expr in SEPARATOR.split(str(exprs).strip()): self.append_expression(expr) @@ -45,7 +45,7 @@ class BaseField: def get_value(self, dateval: datetime) -> int: return getattr(dateval, self.name) - def get_next_value(self, dateval: datetime) -> Optional[int]: + def get_next_value(self, dateval: datetime) -> int | None: smallest = None for expr in self.expressions: value = expr.get_next_value(dateval, self) diff --git a/src/apscheduler/triggers/date.py b/src/apscheduler/triggers/date.py index 08494c3..173c972 100644 --- a/src/apscheduler/triggers/date.py +++ b/src/apscheduler/triggers/date.py @@ -1,7 +1,7 @@ from __future__ import annotations from datetime import datetime -from typing import Any, Optional +from typing import Any import attrs @@ -21,7 +21,7 @@ class DateTrigger(Trigger): run_time: datetime = attrs.field(converter=as_aware_datetime) _completed: bool = attrs.field(init=False, eq=False, default=False) - def next(self) -> Optional[datetime]: + def next(self) -> datetime | None: if not self._completed: self._completed = True return self.run_time diff --git a/src/apscheduler/triggers/interval.py b/src/apscheduler/triggers/interval.py index 0f85d36..629499b 100644 --- a/src/apscheduler/triggers/interval.py +++ b/src/apscheduler/triggers/interval.py @@ -1,7 +1,7 @@ from __future__ import annotations from datetime import datetime, timedelta -from typing import Any, Optional +from typing import Any import attrs @@ -37,9 +37,9 @@ class IntervalTrigger(Trigger): seconds: float = 0 microseconds: float = 0 start_time: datetime = attrs.field(converter=as_aware_datetime, factory=datetime.now) - end_time: Optional[datetime] = attrs.field(converter=as_aware_datetime, default=None) + end_time: datetime | None = attrs.field(converter=as_aware_datetime, default=None) _interval: timedelta = attrs.field(init=False, eq=False, repr=False) - _last_fire_time: Optional[datetime] = attrs.field(init=False, eq=False, default=None) + _last_fire_time: datetime | None = attrs.field(init=False, eq=False, default=None) def __attrs_post_init__(self) -> None: self._interval = timedelta(weeks=self.weeks, days=self.days, hours=self.hours, @@ -52,7 +52,7 @@ class IntervalTrigger(Trigger): if self.end_time and self.end_time < self.start_time: raise ValueError('end_time cannot be earlier than start_time') - def next(self) -> Optional[datetime]: + def next(self) -> datetime | None: if self._last_fire_time is None: self._last_fire_time = self.start_time else: |