Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/bones/date.py: 42%
147 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-09-03 13:41 +0000
1from datetime import datetime, timedelta, timezone
2import typing as t
4import pytz
5import tzlocal
7from viur.core import conf, current, db
8from viur.core.bones.base import BaseBone, ReadFromClientError, ReadFromClientErrorSeverity
9from viur.core.utils import utcNow
12class DateBone(BaseBone):
13 """
14 DateBone is a bone that can handle date and/or time information. It can store date and time information
15 separately, as well as localize the time based on the user's timezone.
17 :param bool creationMagic: Use the current time as value when creating an entity; ignoring this bone if the
18 entity gets updated.
19 :param bool updateMagic: Use the current time whenever this entity is saved.
20 :param bool date: If True, the bone will contain date information.
21 :param time: If True, the bone will contain time information.
22 :param localize: If True, the user's timezone is assumed for input and output. This is only valid if both 'date'
23 and 'time' are set to True. By default, UTC time is used.
24 """
25 # FIXME: the class has no parameters; merge with __init__
26 type = "date"
28 def __init__(
29 self,
30 *,
31 creationMagic: bool = False,
32 date: bool = True,
33 localize: bool = None,
34 naive: bool = False,
35 time: bool = True,
36 updateMagic: bool = False,
37 **kwargs
38 ):
39 """
40 Initializes a new DateBone.
42 :param creationMagic: Use the current time as value when creating an entity; ignoring this bone if the
43 entity gets updated.
44 :param updateMagic: Use the current time whenever this entity is saved.
45 :param date: Should this bone contain a date-information?
46 :param time: Should this bone contain time information?
47 :param localize: Assume users timezone for in and output? Only valid if this bone
48 contains date and time-information! Per default, UTC time is used.
49 :param naive: Use naive datetime for this bone, the default is aware.
50 """
51 super().__init__(**kwargs)
53 # Either date or time must be set
54 if not (date or time): 54 ↛ 55line 54 didn't jump to line 55 because the condition on line 54 was never true
55 raise ValueError("Attempt to create an empty DateBone! Set date or time to True!")
57 # Localize-flag only possible with date and time
58 if localize and not (date and time): 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true
59 raise ValueError("Localization is only possible with date and time!")
60 # Default localize all DateBones, if not explicitly defined
61 elif localize is None and not naive: 61 ↛ 64line 61 didn't jump to line 64 because the condition on line 61 was always true
62 localize = date and time
64 if naive and localize: 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true
65 raise ValueError("Localize and naive is not possible!")
67 # Magic is only possible in non-multiple bones and why ever only on readonly bones...
68 if creationMagic or updateMagic: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true
69 if self.multiple:
70 raise ValueError("Cannot be multiple and have a creation/update-magic set!")
72 self.readonly = True # todo: why???
74 self.creationMagic = creationMagic
75 self.updateMagic = updateMagic
76 self.date = date
77 self.time = time
78 self.localize = localize
79 self.naive = naive
81 def singleValueFromClient(self, value, skel, bone_name, client_data):
82 """
83 Reads a value from the client. If the value is valid for this bone, it stores the value and returns None.
84 Otherwise, the previous value is left unchanged, and an error message is returned.
85 The value is assumed to be in the local time zone only if both self.date and self.time are set to True and
86 self.localize is True.
87 **Value is valid if, when converted into String, it complies following formats:**
88 is digit (may include one '-') and valid POSIX timestamp: converted from timestamp;
89 assumes UTC timezone
90 is digit (may include one '-') and NOT valid POSIX timestamp and not date and time: interpreted as
91 seconds after epoch
92 'now': current time
93 'nowX', where X converted into String is added as seconds to current time
94 '%H:%M:%S' if not date and time
95 '%M:%S' if not date and time
96 '%S' if not date and time
97 '%Y-%m-%d %H:%M:%S' (ISO date format)
98 '%Y-%m-%d %H:%M' (ISO date format)
99 '%Y-%m-%d' (ISO date format)
100 '%m/%d/%Y %H:%M:%S' (US date-format)
101 '%m/%d/%Y %H:%M' (US date-format)
102 '%m/%d/%Y' (US date-format)
103 '%d.%m.%Y %H:%M:%S' (EU date-format)
104 '%d.%m.%Y %H:%M' (EU date-format)
105 '%d.%m.%Y' (EU date-format)
107 The resulting year must be >= 1900.
109 :param bone_name: Our name in the skeleton
110 :param client_data: *User-supplied* request-data, has to be of valid format
111 :returns: tuple[datetime or None, [Errors] or None]
112 """
113 time_zone = self.guessTimeZone()
114 value = str(value) # always enforce value to be a str
116 if value.replace("-", "", 1).replace(".", "", 1).isdigit(): 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 if int(value) < -1 * (2 ** 30) or int(value) > (2 ** 31) - 2:
118 value = None
119 else:
120 value = datetime.fromtimestamp(float(value), tz=time_zone).replace(microsecond=0)
122 elif not self.date and self.time: 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true
123 try:
124 value = datetime.fromisoformat(value)
126 except ValueError:
127 try:
128 if value.count(":") > 1:
129 (hour, minute, second) = [int(x.strip()) for x in value.split(":")]
130 value = datetime(year=1970, month=1, day=1, hour=hour, minute=minute, second=second,
131 tzinfo=time_zone)
132 elif value.count(":") > 0:
133 (hour, minute) = [int(x.strip()) for x in value.split(":")]
134 value = datetime(year=1970, month=1, day=1, hour=hour, minute=minute, tzinfo=time_zone)
135 elif value.replace("-", "", 1).isdigit():
136 value = datetime(year=1970, month=1, day=1, second=int(value), tzinfo=time_zone)
137 else:
138 value = None
140 except ValueError:
141 value = None
143 elif value.lower().startswith("now"):
144 now = datetime.now(time_zone)
145 if len(value) > 4:
146 try:
147 now += timedelta(seconds=int(value[3:]))
148 except ValueError:
149 now = None
151 value = now
153 else:
154 # try to parse ISO-formatted date string
155 try:
156 value = datetime.fromisoformat(value)
157 except ValueError:
158 # otherwise, test against several format strings
159 for fmt in ( 159 ↛ 176line 159 didn't jump to line 176 because the loop on line 159 didn't complete
160 "%Y-%m-%d %H:%M:%S",
161 "%m/%d/%Y %H:%M:%S",
162 "%d.%m.%Y %H:%M:%S",
163 "%Y-%m-%d %H:%M",
164 "%m/%d/%Y %H:%M",
165 "%d.%m.%Y %H:%M",
166 "%Y-%m-%d",
167 "%m/%d/%Y",
168 "%d.%m.%Y",
169 ):
170 try:
171 value = datetime.strptime(value, fmt)
172 break
173 except ValueError:
174 continue
175 else:
176 value = None
178 if not value:
179 return self.getEmptyValue(), [
180 ReadFromClientError(ReadFromClientErrorSeverity.Invalid, "Invalid value entered")
181 ]
183 if value.tzinfo and self.naive: 183 ↛ 184line 183 didn't jump to line 184 because the condition on line 183 was never true
184 return self.getEmptyValue(), [
185 ReadFromClientError(ReadFromClientErrorSeverity.Invalid, "Datetime must be naive")
186 ]
188 if not value.tzinfo and not self.naive:
189 value = time_zone.localize(value)
191 # remove microseconds
192 # TODO: might become configurable
193 value = value.replace(microsecond=0)
195 if err := self.isInvalid(value): 195 ↛ 196line 195 didn't jump to line 196 because the condition on line 195 was never true
196 return self.getEmptyValue(), [ReadFromClientError(ReadFromClientErrorSeverity.Invalid, err)]
198 return value, None
200 def isInvalid(self, value):
201 """
202 Validates the input value to ensure that the year is greater than or equal to 1900. If the year is less
203 than 1900, it returns an error message. Otherwise, it calls the superclass's isInvalid method to perform
204 any additional validations.
206 This check is important because the strftime function, which is used to format dates in Python, will
207 break if the year is less than 1900.
209 :param datetime value: The input value to be validated, expected to be a datetime object.
211 :returns: An error message if the year is less than 1900, otherwise the result of calling
212 the superclass's isInvalid method.
213 :rtype: str or None
214 """
215 if isinstance(value, datetime): 215 ↛ 219line 215 didn't jump to line 219 because the condition on line 215 was always true
216 if value.year < 1900: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 return "Year must be >= 1900"
219 return super().isInvalid(value)
221 def guessTimeZone(self):
222 """
223 Tries to guess the user's time zone based on request headers. If the time zone cannot be guessed, it
224 falls back to using the UTC time zone. The guessed time zone is then cached for future use during the
225 current request.
227 :returns: The guessed time zone for the user or a default time zone (UTC) if the time zone cannot be guessed.
228 :rtype: pytz timezone object
229 """
230 if self.naive: 230 ↛ 231line 230 didn't jump to line 231 because the condition on line 230 was never true
231 return None
232 if not (self.date and self.time and self.localize): 232 ↛ 233line 232 didn't jump to line 233 because the condition on line 232 was never true
233 return pytz.utc
235 if conf.instance.is_dev_server: 235 ↛ 236line 235 didn't jump to line 236 because the condition on line 235 was never true
236 return pytz.timezone(tzlocal.get_localzone_name())
238 timeZone = pytz.utc # Default fallback
239 currReqData = current.request_data.get()
241 try:
242 # Check the local cache first
243 if "timeZone" in currReqData: 243 ↛ 244, 243 ↛ 2452 missed branches: 1) line 243 didn't jump to line 244 because the condition on line 243 was never true, 2) line 243 didn't jump to line 245 because the condition on line 243 was always true
244 return currReqData["timeZone"]
245 headers = current.request.get().request.headers
246 if "X-Appengine-Country" in headers:
247 country = headers["X-Appengine-Country"]
248 else: # Maybe local development Server - no way to guess it here
249 return timeZone
250 tzList = pytz.country_timezones[country]
251 except: # Non-User generated request (deferred call; task queue etc), or no pytz
252 return timeZone
253 if len(tzList) == 1: # Fine - the country has exactly one timezone
254 timeZone = pytz.timezone(tzList[0])
255 elif country.lower() == "us": # Fallback for the US
256 timeZone = pytz.timezone("EST")
257 elif country.lower() == "de": # For some freaking reason Germany is listed with two timezones
258 timeZone = pytz.timezone("Europe/Berlin")
259 elif country.lower() == "au":
260 timeZone = pytz.timezone("Australia/Canberra") # Equivalent to NSW/Sydney :)
261 else: # The user is in a Country which has more than one timezone
262 pass
263 currReqData["timeZone"] = timeZone # Cache the result
264 return timeZone
266 def singleValueSerialize(self, value, skel: 'SkeletonInstance', name: str, parentIndexed: bool):
267 """
268 Prepares a single value for storage by removing any unwanted parts of the datetime object, such as
269 microseconds or adjusting the date and time components depending on the configuration of the dateBone.
270 The method also ensures that the datetime object is timezone aware.
272 :param datetime value: The input datetime value to be serialized.
273 :param SkeletonInstance skel: The instance of the skeleton that contains this bone.
274 :param str name: The name of the bone in the skeleton.
275 :param bool parentIndexed: A boolean indicating if the parent bone is indexed.
276 :returns: The serialized datetime value with unwanted parts removed and timezone-aware.
277 :rtype: datetime
278 """
279 if value:
280 # Crop unwanted values to zero
281 value = value.replace(microsecond=0)
282 if not self.time:
283 value = value.replace(hour=0, minute=0, second=0)
284 elif not self.date:
285 value = value.replace(year=1970, month=1, day=1)
286 if self.naive:
287 value = value.replace(tzinfo=timezone.utc)
288 # We should always deal with timezone aware datetimes
289 assert value.tzinfo, f"Encountered a naive Datetime object in {name} - refusing to save."
290 return value
292 def singleValueUnserialize(self, value):
293 """
294 Converts the serialized datetime value back to its original form. If the datetime object is timezone aware,
295 it adjusts the timezone based on the configuration of the dateBone.
297 :param datetime value: The input serialized datetime value to be unserialized.
298 :returns: The unserialized datetime value with the appropriate timezone applied or None if the input
299 value is not a valid datetime object.
300 :rtype: datetime or None
301 """
302 if isinstance(value, datetime):
303 # Serialized value is timezone aware.
304 if self.naive:
305 value = value.replace(tzinfo=None)
306 return value
307 else:
308 # If local timezone is needed, set here, else force UTC.
309 time_zone = self.guessTimeZone()
310 return value.astimezone(time_zone)
311 else:
312 # We got garbage from the datastore
313 return None
315 def buildDBFilter(self,
316 name: str,
317 skel: 'viur.core.skeleton.SkeletonInstance',
318 dbFilter: db.Query,
319 rawFilter: dict,
320 prefix: t.Optional[str] = None) -> db.Query:
321 """
322 Constructs a datastore filter for date and/or time values based on the given raw filter. It parses the
323 raw filter and, if successful, applies it to the datastore query.
325 :param str name: The name of the dateBone in the skeleton.
326 :param SkeletonInstance skel: The skeleton instance containing the dateBone.
327 :param db.Query dbFilter: The datastore query to which the filter will be applied.
328 :param Dict rawFilter: The raw filter dictionary containing the filter values.
329 :param Optional[str] prefix: An optional prefix to use for the filter key, defaults to None.
330 :returns: The datastore query with the constructed filter applied.
331 :rtype: db.Query
332 """
333 for key in [x for x in rawFilter.keys() if x.startswith(name)]:
334 resDict = {}
335 if not self.fromClient(resDict, key, rawFilter): # Parsing succeeded
336 super().buildDBFilter(name, skel, dbFilter, {key: resDict[key]}, prefix=prefix)
338 return dbFilter
340 def performMagic(self, valuesCache, name, isAdd):
341 """
342 Automatically sets the current date and/or time for a dateBone when a new entry is created or an
343 existing entry is updated, depending on the configuration of creationMagic and updateMagic.
345 :param dict valuesCache: The cache of values to be stored in the datastore.
346 :param str name: The name of the dateBone in the skeleton.
347 :param bool isAdd: A flag indicating whether the operation is adding a new entry (True) or updating an
348 existing one (False).
349 """
350 if (self.creationMagic and isAdd) or self.updateMagic:
351 if self.naive:
352 valuesCache[name] = utcNow().replace(microsecond=0, tzinfo=None)
353 else:
354 valuesCache[name] = utcNow().replace(microsecond=0).astimezone(self.guessTimeZone())
356 def structure(self) -> dict:
357 return super().structure() | {
358 "date": self.date,
359 "time": self.time,
360 "naive": self.naive
361 }