Add timestamp as option to DateTime Fields

AUTHORS.rst

@@ -170,3 +170,4 @@ Contributors (chronological)
 - Isira Seneviratne `@Isira-Seneviratne <https://github.com/Isira-Seneviratne>`_
 - Karthikeyan Singaravelan `@tirkarthi  <https://github.com/tirkarthi>`_
 - Marco Satti `@marcosatti  <https://github.com/marcosatti>`_
+- Ivo Reumkens `@vanHoi <https://github.com/vanHoi>`_

src/marshmallow/fields.py

@@ -1218,25 +1218,32 @@ class DateTime(Field):
     Example: ``'2014-12-22T03:12:58.019077+00:00'``
 
     :param format: Either ``"rfc"`` (for RFC822), ``"iso"`` (for ISO8601),
-        or a date format string. If `None`, defaults to "iso".
+        ``"timestamp"``, ``"timestamp_ms"`` (for a POSIX timestamp) or a date format string.
+        If `None`, defaults to "iso".
     :param kwargs: The same keyword arguments that :class:`Field` receives.
 
     .. versionchanged:: 3.0.0rc9
         Does not modify timezone information on (de)serialization.
+    .. versionchanged:: 3.18
+        Add timestamp as a format.
     """
 
     SERIALIZATION_FUNCS = {
         "iso": utils.isoformat,
         "iso8601": utils.isoformat,
         "rfc": utils.rfcformat,
         "rfc822": utils.rfcformat,
-    }  # type: typing.Dict[str, typing.Callable[[typing.Any], str]]
+        "timestamp": utils.timestamp,
+        "timestamp_ms": utils.timestamp_ms,
+    }  # type: typing.Dict[str, typing.Callable[[typing.Any], str | float]]
 
     DESERIALIZATION_FUNCS = {
         "iso": utils.from_iso_datetime,
         "iso8601": utils.from_iso_datetime,
         "rfc": utils.from_rfc,
         "rfc822": utils.from_rfc,
+        "timestamp": utils.from_timestamp,
+        "timestamp_ms": utils.from_timestamp_ms,
     }  # type: typing.Dict[str, typing.Callable[[str], typing.Any]]
 
     DEFAULT_FORMAT = "iso"
@@ -1252,7 +1259,7 @@ class DateTime(Field):
         "format": '"{input}" cannot be formatted as a {obj_type}.',
     }
 
-    def __init__(self, format: str | None = None, **kwargs):
+    def __init__(self, format: str | None = None, **kwargs) -> None:
         super().__init__(**kwargs)
         # Allow this to be None. It may be set later in the ``_serialize``
         # or ``_deserialize`` methods. This allows a Schema to dynamically set the
@@ -1267,7 +1274,7 @@ def _bind_to_schema(self, field_name, schema):
             or self.DEFAULT_FORMAT
         )
 
-    def _serialize(self, value, attr, obj, **kwargs):
+    def _serialize(self, value, attr, obj, **kwargs) -> str | float | None:
         if value is None:
             return None
         data_format = self.format or self.DEFAULT_FORMAT
@@ -1277,7 +1284,7 @@ def _serialize(self, value, attr, obj, **kwargs):
         else:
             return value.strftime(data_format)
 
-    def _deserialize(self, value, attr, data, **kwargs):
+    def _deserialize(self, value, attr, data, **kwargs) -> dt.datetime:
         if not value:  # Falsy values, e.g. '', None, [] are not valid
             raise self.make_error("invalid", input=value, obj_type=self.OBJ_TYPE)
         data_format = self.format or self.DEFAULT_FORMAT
@@ -1298,7 +1305,7 @@ def _deserialize(self, value, attr, data, **kwargs):
                 ) from error
 
     @staticmethod
-    def _make_object_from_format(value, data_format):
+    def _make_object_from_format(value, data_format) -> dt.datetime:
         return dt.datetime.strptime(value, data_format)
 
 
@@ -1323,11 +1330,11 @@ def __init__(
         *,
         timezone: dt.timezone | None = None,
         **kwargs,
-    ):
+    ) -> None:
         super().__init__(format=format, **kwargs)
         self.timezone = timezone
 
-    def _deserialize(self, value, attr, data, **kwargs):
+    def _deserialize(self, value, attr, data, **kwargs) -> dt.datetime:
         ret = super()._deserialize(value, attr, data, **kwargs)
         if is_aware(ret):
             if self.timezone is None:
@@ -1360,11 +1367,11 @@ def __init__(
         *,
         default_timezone: dt.tzinfo | None = None,
         **kwargs,
-    ):
+    ) -> None:
         super().__init__(format=format, **kwargs)
         self.default_timezone = default_timezone
 
-    def _deserialize(self, value, attr, data, **kwargs):
+    def _deserialize(self, value, attr, data, **kwargs) -> dt.datetime:
         ret = super()._deserialize(value, attr, data, **kwargs)
         if not is_aware(ret):
             if self.default_timezone is None:

src/marshmallow/utils.py

@@ -190,6 +190,36 @@ def from_iso_date(value):
     return dt.date(**kw)
 
 
+def from_timestamp(value: typing.Any) -> dt.datetime:
+    value = float(value)
+    if value < 0:
+        raise ValueError("Not a valid POSIX timestamp")
+
+    # Load a timestamp with utc as timezone to prevent using system timezone.
+    # Then set timezone to None, to let the Field handle adding timezone info.
+    return dt.datetime.fromtimestamp(value, tz=dt.timezone.utc).replace(tzinfo=None)
+
+
+def from_timestamp_ms(value: typing.Any) -> dt.datetime:
+    value = float(value)
+    return from_timestamp(value / 1000)
+
+
+def timestamp(
+    value: dt.datetime,
+) -> float:
+    if is_aware(value):
+        return value.timestamp()
+
+    # When a date is naive, use utc as zone info to prevent using system timezone.
+    # See Python docs for more info: https://docs.python.org/3.10/library/datetime.html#datetime.datetime.timestamp
+    return value.replace(tzinfo=dt.timezone.utc).timestamp()
+
+
+def timestamp_ms(value: dt.datetime) -> float:
+    return timestamp(value) * 1000
+
+
 def isoformat(datetime: dt.datetime) -> str:
     """Return the ISO8601-formatted representation of a datetime object.
 

tests/test_deserialization.py

@@ -518,6 +518,50 @@ def test_iso_datetime_field_deserialization(self, fmt, value, expected, aware):
         else:
             assert field.deserialize(value) == expected
 
+    @pytest.mark.parametrize(
+        ("fmt", "value", "expected"),
+        [
+            ("timestamp", 1384043025, dt.datetime(2013, 11, 10, 0, 23, 45)),
+            ("timestamp", "1384043025", dt.datetime(2013, 11, 10, 0, 23, 45)),
+            ("timestamp", 1384043025, dt.datetime(2013, 11, 10, 0, 23, 45)),
+            ("timestamp", 1384043025.12, dt.datetime(2013, 11, 10, 0, 23, 45, 120000)),
+            (
+                "timestamp",
+                1384043025.123456,
+                dt.datetime(2013, 11, 10, 0, 23, 45, 123456),
+            ),
+            ("timestamp", 1, dt.datetime(1970, 1, 1, 0, 0, 1)),
+            ("timestamp_ms", 1384043025000, dt.datetime(2013, 11, 10, 0, 23, 45)),
+            ("timestamp_ms", 1000, dt.datetime(1970, 1, 1, 0, 0, 1)),
+        ],
+    )
+    def test_timestamp_field_deserialization(self, fmt, value, expected):
+        field = fields.DateTime(format=fmt)
+        assert field.deserialize(value) == expected
+
+        # By default, a datetime from a timestamp is never aware.
+        field = fields.NaiveDateTime(format=fmt)
+        assert field.deserialize(value) == expected
+
+        field = fields.AwareDateTime(format=fmt)
+        with pytest.raises(ValidationError, match="Not a valid aware datetime."):
+            field.deserialize(value)
+
+        # But it can be added by providing a default.
+        field = fields.AwareDateTime(format=fmt, default_timezone=central)
+        expected_aware = expected.replace(tzinfo=central)
+        assert field.deserialize(value) == expected_aware
+
+    @pytest.mark.parametrize("fmt", ["timestamp", "timestamp_ms"])
+    @pytest.mark.parametrize(
+        "in_value",
+        ["", "!@#", 0, -1, dt.datetime(2013, 11, 10, 1, 23, 45)],
+    )
+    def test_invalid_timestamp_field_deserialization(self, fmt, in_value):
+        field = fields.DateTime(format="timestamp")
+        with pytest.raises(ValidationError, match="Not a valid datetime."):
+            field.deserialize(in_value)
+
     @pytest.mark.parametrize(
         ("fmt", "timezone", "value", "expected"),
         [

tests/test_serialization.py

@@ -554,6 +554,38 @@ def test_datetime_field_rfc822(self, fmt, value, expected):
         field = fields.DateTime(format=fmt)
         assert field.serialize("d", {"d": value}) == expected
 
+    @pytest.mark.parametrize(
+        ("fmt", "value", "expected"),
+        [
+            ("timestamp", dt.datetime(1970, 1, 1), 0),
+            ("timestamp", dt.datetime(2013, 11, 10, 0, 23, 45), 1384043025),
+            (
+                "timestamp",
+                dt.datetime(2013, 11, 10, 0, 23, 45, tzinfo=dt.timezone.utc),
+                1384043025,
+            ),
+            (
+                "timestamp",
+                central.localize(dt.datetime(2013, 11, 10, 0, 23, 45), is_dst=False),
+                1384064625,
+            ),
+            ("timestamp_ms", dt.datetime(2013, 11, 10, 0, 23, 45), 1384043025000),
+            (
+                "timestamp_ms",
+                dt.datetime(2013, 11, 10, 0, 23, 45, tzinfo=dt.timezone.utc),
+                1384043025000,
+            ),
+            (
+                "timestamp_ms",
+                central.localize(dt.datetime(2013, 11, 10, 0, 23, 45), is_dst=False),
+                1384064625000,
+            ),
+        ],
+    )
+    def test_datetime_field_timestamp(self, fmt, value, expected):
+        field = fields.DateTime(format=fmt)
+        assert field.serialize("d", {"d": value}) == expected
+
     @pytest.mark.parametrize("fmt", ["iso", "iso8601", None])
     @pytest.mark.parametrize(
         ("value", "expected"),