summaryrefslogtreecommitdiff
path: root/src/apscheduler/triggers
diff options
context:
space:
mode:
Diffstat (limited to 'src/apscheduler/triggers')
-rw-r--r--src/apscheduler/triggers/calendarinterval.py6
-rw-r--r--src/apscheduler/triggers/combining.py20
-rw-r--r--src/apscheduler/triggers/cron/__init__.py10
-rw-r--r--src/apscheduler/triggers/cron/expressions.py7
-rw-r--r--src/apscheduler/triggers/cron/fields.py6
-rw-r--r--src/apscheduler/triggers/date.py4
-rw-r--r--src/apscheduler/triggers/interval.py8
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: