Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
The table of contents is too big for display.
Diff view
Diff view
  •  
  •  
  •  
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,29 @@ def _serialize_bytes(o, format: typing.Optional[str] = None) -> str:
return encoded


def _serialize_duration(td: timedelta, format: typing.Optional[str] = None):
"""Serialize a timedelta to its wire representation.

For the ``seconds``/``milliseconds`` encodings the value is converted to a
numeric value, otherwise it falls back to an ISO 8601 duration string.

:param timedelta td: The timedelta to serialize.
:param str format: The duration encoding format.
:rtype: int or float or str
:return: serialized duration
"""
seconds = td.total_seconds()
if format == "duration-seconds-int":
return int(seconds)
if format == "duration-seconds-float":
return seconds
if format == "duration-milliseconds-int":
return int(seconds * 1000)
if format == "duration-milliseconds-float":
return seconds * 1000
return _timedelta_as_isostr(td)


def _serialize_datetime(o, format: typing.Optional[str] = None):
if hasattr(o, "year") and hasattr(o, "hour"):
if format == "rfc7231":
Expand Down Expand Up @@ -301,6 +324,12 @@ def _deserialize_duration(attr):
return isodate.parse_duration(attr)


def _deserialize_duration_numeric(attr, unit):
if isinstance(attr, timedelta):
return attr
return timedelta(**{unit: float(attr)})


def _deserialize_decimal(attr):
if isinstance(attr, decimal.Decimal):
return attr
Expand Down Expand Up @@ -330,6 +359,10 @@ def _deserialize_int_as_str(attr):
"unix-timestamp": _deserialize_datetime_unix_timestamp,
"base64": _deserialize_bytes,
"base64url": _deserialize_bytes_base64,
"duration-seconds-int": functools.partial(_deserialize_duration_numeric, unit="seconds"),
"duration-seconds-float": functools.partial(_deserialize_duration_numeric, unit="seconds"),
"duration-milliseconds-int": functools.partial(_deserialize_duration_numeric, unit="milliseconds"),
"duration-milliseconds-float": functools.partial(_deserialize_duration_numeric, unit="milliseconds"),
}


Expand Down Expand Up @@ -564,7 +597,7 @@ def _serialize(o, format: typing.Optional[str] = None): # pylint: disable=too-m
pass
# Last, try datetime.timedelta
try:
return _timedelta_as_isostr(o)
return _serialize_duration(o, format)
except AttributeError:
# This will be raised when it hits value.total_seconds in the method above
pass
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -520,6 +520,10 @@ def __init__(self, classes: Optional[Mapping[str, type]] = None) -> None:
"rfc-1123": Serializer.serialize_rfc,
"unix-time": Serializer.serialize_unix,
"duration": Serializer.serialize_duration,
"duration-seconds-int": Serializer.serialize_duration_seconds_int,
"duration-seconds-float": Serializer.serialize_duration_seconds_float,
"duration-milliseconds-int": Serializer.serialize_duration_milliseconds_int,
"duration-milliseconds-float": Serializer.serialize_duration_milliseconds_float,
"date": Serializer.serialize_date,
"time": Serializer.serialize_time,
"decimal": Serializer.serialize_decimal,
Expand Down Expand Up @@ -1109,6 +1113,61 @@ def serialize_duration(attr, **kwargs): # pylint: disable=unused-argument
attr = isodate.parse_duration(attr)
return isodate.duration_isoformat(attr)

@staticmethod
def _serialize_duration_numeric(attr, scale, as_int):
"""Serialize a TimeDelta into a numeric value scaled to the wire unit.

:param TimeDelta attr: Object to be serialized.
:param int scale: Multiplier applied to total seconds (1 for seconds, 1000 for milliseconds).
:param bool as_int: Whether to truncate the result to an int.
:rtype: int or float
:return: serialized duration
"""
if isinstance(attr, str):
attr = isodate.parse_duration(attr)
value = attr.total_seconds() * scale if isinstance(attr, datetime.timedelta) else attr
return int(value) if as_int else float(value)

@staticmethod
def serialize_duration_seconds_int(attr, **kwargs): # pylint: disable=unused-argument
"""Serialize TimeDelta object into an integer number of seconds.

:param TimeDelta attr: Object to be serialized.
:rtype: int
:return: serialized duration
"""
return Serializer._serialize_duration_numeric(attr, 1, True)

@staticmethod
def serialize_duration_seconds_float(attr, **kwargs): # pylint: disable=unused-argument
"""Serialize TimeDelta object into a floating point number of seconds.

:param TimeDelta attr: Object to be serialized.
:rtype: float
:return: serialized duration
"""
return Serializer._serialize_duration_numeric(attr, 1, False)

@staticmethod
def serialize_duration_milliseconds_int(attr, **kwargs): # pylint: disable=unused-argument
"""Serialize TimeDelta object into an integer number of milliseconds.

:param TimeDelta attr: Object to be serialized.
:rtype: int
:return: serialized duration
"""
return Serializer._serialize_duration_numeric(attr, 1000, True)

@staticmethod
def serialize_duration_milliseconds_float(attr, **kwargs): # pylint: disable=unused-argument
"""Serialize TimeDelta object into a floating point number of milliseconds.

:param TimeDelta attr: Object to be serialized.
:rtype: float
:return: serialized duration
"""
return Serializer._serialize_duration_numeric(attr, 1000, False)

@staticmethod
def serialize_rfc(attr, **kwargs): # pylint: disable=unused-argument
"""Serialize Datetime object into RFC-1123 formatted string.
Expand Down Expand Up @@ -1381,6 +1440,10 @@ def __init__(self, classes: Optional[Mapping[str, type]] = None) -> None:
"rfc-1123": Deserializer.deserialize_rfc,
"unix-time": Deserializer.deserialize_unix,
"duration": Deserializer.deserialize_duration,
"duration-seconds-int": Deserializer.deserialize_duration_seconds,
"duration-seconds-float": Deserializer.deserialize_duration_seconds,
"duration-milliseconds-int": Deserializer.deserialize_duration_milliseconds,
"duration-milliseconds-float": Deserializer.deserialize_duration_milliseconds,
"date": Deserializer.deserialize_date,
"time": Deserializer.deserialize_time,
"decimal": Deserializer.deserialize_decimal,
Expand All @@ -1393,6 +1456,10 @@ def __init__(self, classes: Optional[Mapping[str, type]] = None) -> None:
}
self.deserialize_expected_types = {
"duration": (isodate.Duration, datetime.timedelta),
"duration-seconds-int": (isodate.Duration, datetime.timedelta),
"duration-seconds-float": (isodate.Duration, datetime.timedelta),
"duration-milliseconds-int": (isodate.Duration, datetime.timedelta),
"duration-milliseconds-float": (isodate.Duration, datetime.timedelta),
"iso-8601": (datetime.datetime),
}
self.dependencies: dict[str, type] = dict(classes) if classes else {}
Expand Down Expand Up @@ -1954,6 +2021,48 @@ def deserialize_duration(attr):
raise DeserializationError(msg) from err
return duration

@staticmethod
def _deserialize_duration_numeric(attr, unit):
"""Deserialize a numeric duration value into a TimeDelta object.

:param float attr: response value to be deserialized.
:param str unit: The wire unit, used as the ``timedelta`` keyword
(``"seconds"`` or ``"milliseconds"``).
:return: Deserialized duration
:rtype: TimeDelta
:raises DeserializationError: if value is invalid.
"""
if isinstance(attr, ET.Element):
attr = attr.text
try:
duration = datetime.timedelta(**{unit: float(attr)}) # type: ignore
except (ValueError, OverflowError, TypeError) as err:
msg = "Cannot deserialize duration object."
raise DeserializationError(msg) from err
return duration

@staticmethod
def deserialize_duration_seconds(attr):
"""Deserialize a numeric number of seconds into a TimeDelta object.

:param float attr: response value to be deserialized.
:return: Deserialized duration
:rtype: TimeDelta
:raises DeserializationError: if value is invalid.
"""
return Deserializer._deserialize_duration_numeric(attr, "seconds")

@staticmethod
def deserialize_duration_milliseconds(attr):
"""Deserialize a numeric number of milliseconds into a TimeDelta object.

:param float attr: response value to be deserialized.
:return: Deserialized duration
:rtype: TimeDelta
:raises DeserializationError: if value is invalid.
"""
return Deserializer._deserialize_duration_numeric(attr, "milliseconds")

@staticmethod
def deserialize_date(attr):
"""Deserialize ISO-8601 formatted string into Date object.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,29 @@ def _serialize_bytes(o, format: typing.Optional[str] = None) -> str:
return encoded


def _serialize_duration(td: timedelta, format: typing.Optional[str] = None):
"""Serialize a timedelta to its wire representation.

For the ``seconds``/``milliseconds`` encodings the value is converted to a
numeric value, otherwise it falls back to an ISO 8601 duration string.

:param timedelta td: The timedelta to serialize.
:param str format: The duration encoding format.
:rtype: int or float or str
:return: serialized duration
"""
seconds = td.total_seconds()
if format == "duration-seconds-int":
return int(seconds)
if format == "duration-seconds-float":
return seconds
if format == "duration-milliseconds-int":
return int(seconds * 1000)
if format == "duration-milliseconds-float":
return seconds * 1000
return _timedelta_as_isostr(td)


def _serialize_datetime(o, format: typing.Optional[str] = None):
if hasattr(o, "year") and hasattr(o, "hour"):
if format == "rfc7231":
Expand Down Expand Up @@ -301,6 +324,12 @@ def _deserialize_duration(attr):
return isodate.parse_duration(attr)


def _deserialize_duration_numeric(attr, unit):
if isinstance(attr, timedelta):
return attr
return timedelta(**{unit: float(attr)})


def _deserialize_decimal(attr):
if isinstance(attr, decimal.Decimal):
return attr
Expand Down Expand Up @@ -330,6 +359,10 @@ def _deserialize_int_as_str(attr):
"unix-timestamp": _deserialize_datetime_unix_timestamp,
"base64": _deserialize_bytes,
"base64url": _deserialize_bytes_base64,
"duration-seconds-int": functools.partial(_deserialize_duration_numeric, unit="seconds"),
"duration-seconds-float": functools.partial(_deserialize_duration_numeric, unit="seconds"),
"duration-milliseconds-int": functools.partial(_deserialize_duration_numeric, unit="milliseconds"),
"duration-milliseconds-float": functools.partial(_deserialize_duration_numeric, unit="milliseconds"),
}


Expand Down Expand Up @@ -564,7 +597,7 @@ def _serialize(o, format: typing.Optional[str] = None): # pylint: disable=too-m
pass
# Last, try datetime.timedelta
try:
return _timedelta_as_isostr(o)
return _serialize_duration(o, format)
except AttributeError:
# This will be raised when it hits value.total_seconds in the method above
pass
Expand Down
Loading