Changeset View
Changeset View
Standalone View
Standalone View
swh/model/model.py
Show All 13 Lines | |||||
All classes define a ``from_dict`` class method and a ``to_dict`` | All classes define a ``from_dict`` class method and a ``to_dict`` | ||||
method to convert between them and msgpack-serializable objects. | method to convert between them and msgpack-serializable objects. | ||||
""" | """ | ||||
from abc import ABCMeta, abstractmethod | from abc import ABCMeta, abstractmethod | ||||
import datetime | import datetime | ||||
from enum import Enum | from enum import Enum | ||||
import hashlib | import hashlib | ||||
import math | |||||
from typing import Any, Dict, Iterable, Optional, Tuple, TypeVar, Union | from typing import Any, Dict, Iterable, Optional, Tuple, TypeVar, Union | ||||
import attr | import attr | ||||
from attrs_strict import AttributeTypeError | from attrs_strict import AttributeTypeError | ||||
import dateutil.parser | import dateutil.parser | ||||
import iso8601 | import iso8601 | ||||
from typing_extensions import Final | from typing_extensions import Final | ||||
▲ Show 20 Lines • Show All 442 Lines • ▼ Show 20 Lines | def from_dict(cls, time_representation: Union[Dict, datetime.datetime, int]): | ||||
if "negative_utc" in time_representation: | if "negative_utc" in time_representation: | ||||
negative_utc = time_representation["negative_utc"] | negative_utc = time_representation["negative_utc"] | ||||
if negative_utc is None: | if negative_utc is None: | ||||
negative_utc = False | negative_utc = False | ||||
elif isinstance(time_representation, datetime.datetime): | elif isinstance(time_representation, datetime.datetime): | ||||
microseconds = time_representation.microsecond | microseconds = time_representation.microsecond | ||||
if microseconds: | if microseconds: | ||||
time_representation = time_representation.replace(microsecond=0) | time_representation = time_representation.replace(microsecond=0) | ||||
seconds = int(time_representation.timestamp()) | seconds = math.floor(time_representation.timestamp()) | ||||
utcoffset = time_representation.utcoffset() | utcoffset = time_representation.utcoffset() | ||||
if utcoffset is None: | if utcoffset is None: | ||||
raise ValueError( | raise ValueError( | ||||
f"TimestampWithTimezone.from_dict received datetime without " | f"TimestampWithTimezone.from_dict received datetime without " | ||||
f"timezone: {time_representation}" | f"timezone: {time_representation}" | ||||
) | ) | ||||
# utcoffset is an integer number of minutes | # utcoffset is an integer number of minutes | ||||
seconds_offset = utcoffset.total_seconds() | seconds_offset = utcoffset.total_seconds() | ||||
offset = int(seconds_offset) // 60 | offset = int(seconds_offset) // 60 | ||||
olasd: Maybe that'd be worth a (separate) warning if the remainder is non-zero... | |||||
Done Inline Actionsindeed, I'll add it later vlorentz: indeed, I'll add it later | |||||
elif isinstance(time_representation, int): | elif isinstance(time_representation, int): | ||||
seconds = time_representation | seconds = time_representation | ||||
microseconds = 0 | microseconds = 0 | ||||
offset = 0 | offset = 0 | ||||
else: | else: | ||||
raise ValueError( | raise ValueError( | ||||
f"TimestampWithTimezone.from_dict received non-integer timestamp: " | f"TimestampWithTimezone.from_dict received non-integer timestamp: " | ||||
f"{time_representation!r}" | f"{time_representation!r}" | ||||
▲ Show 20 Lines • Show All 948 Lines • Show Last 20 Lines |
Maybe that'd be worth a (separate) warning if the remainder is non-zero...