diff options
author | Alex Grönholm <alex.gronholm@nextday.fi> | 2021-09-08 11:32:45 +0300 |
---|---|---|
committer | Alex Grönholm <alex.gronholm@nextday.fi> | 2021-09-08 11:32:45 +0300 |
commit | c4e9a6676df302a8d783ed540a2d53682af60a7e (patch) | |
tree | 4d1bcfa1b55a60a1887c4e4127306fcd8c736fa6 /src/apscheduler/triggers | |
parent | 48a5b0eea05f21b4cd21e9305e5c4ab755c88a94 (diff) | |
download | apscheduler-c4e9a6676df302a8d783ed540a2d53682af60a7e.tar.gz |
Converted more classes to use attrs
Diffstat (limited to 'src/apscheduler/triggers')
-rw-r--r-- | src/apscheduler/triggers/calendarinterval.py | 35 | ||||
-rw-r--r-- | src/apscheduler/triggers/combining.py | 27 | ||||
-rw-r--r-- | src/apscheduler/triggers/cron/__init__.py | 41 | ||||
-rw-r--r-- | src/apscheduler/triggers/date.py | 17 | ||||
-rw-r--r-- | src/apscheduler/triggers/interval.py | 39 |
5 files changed, 78 insertions, 81 deletions
diff --git a/src/apscheduler/triggers/calendarinterval.py b/src/apscheduler/triggers/calendarinterval.py index fe4e9f6..c184943 100644 --- a/src/apscheduler/triggers/calendarinterval.py +++ b/src/apscheduler/triggers/calendarinterval.py @@ -3,12 +3,15 @@ from __future__ import annotations from datetime import date, datetime, time, timedelta, tzinfo from typing import Optional +import attr + from ..abc import Trigger from ..marshalling import marshal_date, marshal_timezone, unmarshal_date, unmarshal_timezone from ..util import timezone_repr from ..validators import as_date, as_timezone, require_state_version +@attr.define(kw_only=True) class CalendarIntervalTrigger(Trigger): """ Runs the task on specified calendar-based intervals always at the same exact time of day. @@ -53,23 +56,21 @@ class CalendarIntervalTrigger(Trigger): :param timezone: time zone to use for calculating the next fire time """ - __slots__ = ('years', 'months', 'weeks', 'days', 'start_date', 'end_date', 'timezone', '_time', - '_last_fire_date') - - def __init__(self, *, years: int = 0, months: int = 0, weeks: int = 0, days: int = 0, - hour: int = 0, minute: int = 0, second: int = 0, - start_date: date | str | None = None, - end_date: date | str | None = None, - timezone: str | tzinfo = 'local'): - self.years = years - self.months = months - self.weeks = weeks - self.days = days - self.timezone = as_timezone(timezone) - self.start_date = as_date(start_date) or datetime.now(self.timezone).date() - self.end_date = as_date(end_date) - self._time = time(hour, minute, second, tzinfo=timezone) - self._last_fire_date: Optional[date] = None + years: int = 0 + months: int = 0 + weeks: int = 0 + days: int = 0 + hour: int = 0 + minute: int = 0 + second: int = 0 + start_date: date = attr.field(converter=as_date, factory=date.today) + end_date: date | None = attr.field(converter=as_date, default=None) + timezone: tzinfo = attr.field(converter=as_timezone, default='local') + _time: time = attr.field(init=False, eq=False) + _last_fire_date: Optional[date] = attr.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) if self.years == self.months == self.weeks == self.days == 0: raise ValueError('interval must be at least 1 day long') diff --git a/src/apscheduler/triggers/combining.py b/src/apscheduler/triggers/combining.py index 2bcaa7e..94cbcbe 100644 --- a/src/apscheduler/triggers/combining.py +++ b/src/apscheduler/triggers/combining.py @@ -2,20 +2,20 @@ from __future__ import annotations from abc import abstractmethod from datetime import datetime, timedelta -from typing import Any, Optional, Sequence +from typing import Any, Optional + +import attr from ..abc import Trigger from ..exceptions import MaxIterationsReached from ..marshalling import marshal_object, unmarshal_object -from ..validators import as_list, as_positive_integer, as_timedelta, require_state_version +from ..validators import as_timedelta, require_state_version +@attr.define class BaseCombiningTrigger(Trigger): - __slots__ = 'triggers', '_next_fire_times' - - def __init__(self, triggers: Sequence[Trigger]): - self.triggers = as_list(triggers, Trigger, 'triggers') - self._next_fire_times: list[Optional[datetime]] = [] + triggers: list[Trigger] + _next_fire_times: list[Optional[datetime]] = attr.field(init=False, eq=False, factory=list) def __getstate__(self) -> dict[str, Any]: return { @@ -30,6 +30,7 @@ class BaseCombiningTrigger(Trigger): self._next_fire_times = state['next_fire_times'] +@attr.define class AndTrigger(BaseCombiningTrigger): """ Fires on times produced by the enclosed triggers whenever the fire times are within the given @@ -49,13 +50,8 @@ class AndTrigger(BaseCombiningTrigger): :param max_iterations: maximum number of iterations of fire time calculations before giving up """ - __slots__ = 'threshold', 'max_iterations' - - def __init__(self, triggers: Sequence[Trigger], *, threshold: float | timedelta = 1, - max_iterations: Optional[int] = 10000): - super().__init__(triggers) - self.threshold = as_timedelta(threshold, 'threshold') - self.max_iterations = as_positive_integer(max_iterations, 'max_iterations') + threshold: timedelta = attr.field(converter=as_timedelta, default=1) + max_iterations: Optional[int] = 10000 def next(self) -> Optional[datetime]: if not self._next_fire_times: @@ -106,6 +102,7 @@ class AndTrigger(BaseCombiningTrigger): f'threshold={self.threshold.total_seconds()}, max_iterations={self.max_iterations})' +@attr.define class OrTrigger(BaseCombiningTrigger): """ Fires on every fire time of every trigger in chronological order. @@ -117,8 +114,6 @@ class OrTrigger(BaseCombiningTrigger): :param triggers: triggers to combine """ - __slots__ = () - def next(self) -> Optional[datetime]: # Fill out the fire times on the first run if not self._next_fire_times: diff --git a/src/apscheduler/triggers/cron/__init__.py b/src/apscheduler/triggers/cron/__init__.py index 89ac676..2b5e45b 100644 --- a/src/apscheduler/triggers/cron/__init__.py +++ b/src/apscheduler/triggers/cron/__init__.py @@ -1,7 +1,10 @@ from __future__ import annotations from datetime import datetime, timedelta, tzinfo -from typing import ClassVar, List, Optional, Sequence, Tuple +from typing import ClassVar, Optional, Sequence + +import attr +from tzlocal import get_localzone from ...abc import Trigger from ...marshalling import marshal_date, marshal_timezone, unmarshal_date, unmarshal_timezone @@ -11,6 +14,7 @@ from .fields import ( DEFAULT_VALUES, BaseField, DayOfMonthField, DayOfWeekField, MonthField, WeekField) +@attr.define(kw_only=True) class CronTrigger(Trigger): """ Triggers when current time matches all specified time constraints, similarly to how the UNIX @@ -32,9 +36,7 @@ class CronTrigger(Trigger): .. note:: The first weekday is always **monday**. """ - __slots__ = 'timezone', 'start_time', 'end_time', '_fields', '_last_fire_time' - - FIELDS_MAP: ClassVar[List] = [ + FIELDS_MAP: ClassVar[list[tuple[str, type[BaseField]]]] = [ ('year', BaseField), ('month', MonthField), ('day', DayOfMonthField), @@ -45,18 +47,23 @@ class CronTrigger(Trigger): ('second', BaseField) ] - def __init__(self, *, year: int | str | None = None, month: int | str | None = None, - day: int | str | None = None, week: int | str | None = None, - day_of_week: int | str | None = None, hour: int | str | None = None, - minute: int | str | None = None, second: int | str | None = None, - start_time: datetime | str | None = None, - end_time: datetime | str | None = None, - timezone: str | tzinfo | None = None): - self.timezone = as_timezone(timezone) - self.start_time = (as_aware_datetime(start_time, self.timezone) - or datetime.now(self.timezone)) - self.end_time = as_aware_datetime(end_time, self.timezone) - self._set_fields([year, month, day, week, day_of_week, hour, minute, second]) + year: int | str | None = None + month: int | str | None = None + day: int | str | None = None + week: int | str | None = None + day_of_week: int | str | None = None + hour: int | str | None = None + minute: int | str | None = None + second: int | str | None = None + start_time: datetime = attr.field(converter=as_aware_datetime, factory=datetime.now) + end_time: datetime | None = None + timezone: tzinfo | str = attr.field(converter=as_timezone, factory=get_localzone) + _fields: list[BaseField] = attr.field(init=False, eq=False, factory=list) + _last_fire_time: Optional[datetime] = attr.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 def _set_fields(self, values: Sequence[int | str | None]) -> None: @@ -91,7 +98,7 @@ class CronTrigger(Trigger): 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: datetime, fieldnum: int) -> Tuple[datetime, int]: + def _increment_field_value(self, dateval: datetime, fieldnum: int) -> tuple[datetime, int]: """ Increments the designated field and resets all less significant fields to their minimum values. diff --git a/src/apscheduler/triggers/date.py b/src/apscheduler/triggers/date.py index c142ca7..6d5fc50 100644 --- a/src/apscheduler/triggers/date.py +++ b/src/apscheduler/triggers/date.py @@ -1,28 +1,25 @@ from __future__ import annotations -from datetime import datetime, tzinfo +from datetime import datetime from typing import Optional +import attr + from ..abc import Trigger from ..marshalling import marshal_date, unmarshal_date -from ..validators import as_aware_datetime, as_timezone, require_state_version +from ..validators import as_aware_datetime, require_state_version +@attr.define class DateTrigger(Trigger): """ Triggers once on the given date/time. :param run_time: the date/time to run the job at - :param timezone: time zone to use to convert ``run_time`` into a timezone aware datetime, if it - isn't already """ - __slots__ = 'run_time', '_completed' - - def __init__(self, run_time: datetime, timezone: tzinfo | str = 'local'): - timezone = as_timezone(timezone) - self.run_time = as_aware_datetime(run_time, timezone) - self._completed = False + run_time: datetime = attr.field(converter=as_aware_datetime) + _completed: bool = attr.field(init=False, eq=False, default=False) def next(self) -> Optional[datetime]: if not self._completed: diff --git a/src/apscheduler/triggers/interval.py b/src/apscheduler/triggers/interval.py index 8e933cf..f9bdf3c 100644 --- a/src/apscheduler/triggers/interval.py +++ b/src/apscheduler/triggers/interval.py @@ -1,13 +1,16 @@ from __future__ import annotations -from datetime import datetime, timedelta, tzinfo +from datetime import datetime, timedelta from typing import Optional +import attr + from ..abc import Trigger from ..marshalling import marshal_date, unmarshal_date -from ..validators import as_aware_datetime, as_timezone, require_state_version +from ..validators import as_aware_datetime, require_state_version +@attr.define(kw_only=True) class IntervalTrigger(Trigger): """ Triggers on specified intervals. @@ -23,31 +26,25 @@ class IntervalTrigger(Trigger): :param minutes: number of minutes to wait :param seconds: number of seconds to wait :param microseconds: number of microseconds to wait - :param start_time: first trigger date/time + :param start_time: first trigger date/time (defaults to current date/time if omitted) :param end_time: latest possible date/time to trigger on - :param timezone: time zone used to make any passed naive datetimes timezone aware """ - __slots__ = ('weeks', 'days', 'hours', 'minutes', 'seconds', 'microseconds', 'start_time', - 'end_time', '_interval', '_last_fire_time') - - def __init__(self, *, weeks: float = 0, days: float = 0, hours: float = 0, minutes: float = 0, - seconds: float = 0, microseconds: float = 0, - start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, - timezone: tzinfo | str = 'local'): - self.weeks = weeks - self.days = days - self.hours = hours - self.minutes = minutes - self.seconds = seconds - self.microseconds = microseconds - timezone = as_timezone(timezone) - self.start_time = as_aware_datetime(start_time or datetime.now(), timezone) - self.end_time = as_aware_datetime(end_time, timezone) + weeks: float = 0 + days: float = 0 + hours: float = 0 + minutes: float = 0 + seconds: float = 0 + microseconds: float = 0 + start_time: datetime = attr.field(converter=as_aware_datetime, factory=datetime.now) + end_time: Optional[datetime] = attr.field(converter=as_aware_datetime, default=None) + _interval: timedelta = attr.field(init=False, eq=False) + _last_fire_time: Optional[datetime] = attr.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, minutes=self.minutes, seconds=self.seconds, microseconds=self.microseconds) - self._last_fire_time = None if self._interval.total_seconds() <= 0: raise ValueError('The time interval must be positive') |