1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
from __future__ import annotations
import sys
from datetime import date, datetime, timedelta, timezone, tzinfo
from typing import Any
from attrs import Attribute
from tzlocal import get_localzone
from ._exceptions import DeserializationError
from .abc import Trigger
if sys.version_info >= (3, 9):
from zoneinfo import ZoneInfo
else:
from backports.zoneinfo import ZoneInfo
def as_int(value) -> int | None:
"""Convert the value into an integer."""
if value is None:
return None
return int(value)
def as_timezone(value: str | tzinfo | None) -> tzinfo:
"""
Convert the value into a tzinfo object.
If ``value`` is ``None`` or ``'local'``, use the local timezone.
:param value: the value to be converted
:return: a timezone object
"""
if value is None or value == "local":
return get_localzone()
elif isinstance(value, str):
return ZoneInfo(value)
elif isinstance(value, tzinfo):
if value is timezone.utc:
return ZoneInfo("UTC")
else:
return value
raise TypeError(
f"Expected tzinfo instance or timezone name, got "
f"{value.__class__.__qualname__} instead"
)
def as_date(value: date | str | None) -> date | None:
"""
Convert the value to a date.
:param value: the value to convert to a date
:return: a date object, or ``None`` if ``None`` was given
"""
if value is None:
return None
elif isinstance(value, str):
return date.fromisoformat(value)
elif isinstance(value, date):
return value
raise TypeError(
f"Expected string or date, got {value.__class__.__qualname__} instead"
)
def as_timestamp(value: datetime | None) -> float | None:
if value is None:
return None
return value.timestamp()
def as_ordinal_date(value: date | None) -> int | None:
if value is None:
return None
return value.toordinal()
def as_aware_datetime(value: datetime | str | None) -> datetime | None:
"""
Convert the value to a timezone aware datetime.
:param value: a datetime, an ISO 8601 representation of a datetime, or ``None``
:param tz: timezone to use for making the datetime timezone aware
:return: a timezone aware datetime, or ``None`` if ``None`` was given
"""
if value is None:
return None
if isinstance(value, str):
if value.upper().endswith("Z"):
value = value[:-1] + "+00:00"
value = datetime.fromisoformat(value)
if isinstance(value, datetime):
if not value.tzinfo:
return value.replace(tzinfo=get_localzone())
else:
return value
raise TypeError(
f"Expected string or datetime, got {value.__class__.__qualname__} instead"
)
def positive_number(instance, attribute, value) -> None:
if value <= 0:
raise ValueError(f"Expected positive number, got {value} instead")
def non_negative_number(instance, attribute, value) -> None:
if value < 0:
raise ValueError(f"Expected non-negative number, got {value} instead")
def as_positive_integer(value, name: str) -> int:
if isinstance(value, int):
if value > 0:
return value
else:
raise ValueError(f"{name} must be positive")
raise TypeError(
f"{name} must be an integer, got {value.__class__.__name__} instead"
)
def as_timedelta(value: timedelta | float) -> timedelta:
if isinstance(value, (int, float)):
return timedelta(seconds=value)
elif isinstance(value, timedelta):
return value
# raise TypeError(f'{attribute.name} must be a timedelta or number of seconds, got '
# f'{value.__class__.__name__} instead')
def as_list(value, element_type: type, name: str) -> list:
value = list(value)
for i, element in enumerate(value):
if not isinstance(element, element_type):
raise TypeError(
f"Element at index {i} of {name} is not of the expected type "
f"({element_type.__name__}"
)
return value
def aware_datetime(instance: Any, attribute: Attribute, value: datetime) -> None:
if not value.tzinfo:
raise ValueError(f"{attribute.name} must be a timezone aware datetime")
def require_state_version(
trigger: Trigger, state: dict[str, Any], max_version: int
) -> None:
try:
if state["version"] > max_version:
raise DeserializationError(
f"{trigger.__class__.__name__} received a serialized state with "
f'version {state["version"]}, but it only supports up to version '
f"{max_version}. This can happen when an older version of APScheduler "
f"is being used with a data store that was previously used with a "
f"newer APScheduler version."
)
except KeyError as exc:
raise DeserializationError(
'Missing "version" key in the serialized state'
) from exc
|
