Epoch - unix timestamp

using type float, format: date-time to serialize datetime.datetime

Author: commonism

pydantic_extra_types/epoch.py

@@ -0,0 +1,35 @@
+import datetime
+from typing import Any, Callable
+
+from pydantic import GetJsonSchemaHandler
+from pydantic.json_schema import JsonSchemaValue
+from pydantic_core import CoreSchema, core_schema
+
+EPOCH = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
+
+
+class Epoch(datetime.datetime):
+    @classmethod
+    def __get_pydantic_json_schema__(
+        cls, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler
+    ) -> JsonSchemaValue:
+        field_schema: dict[str, Any] = {}
+        field_schema.update(type='number', format='date-time')
+        return field_schema
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls, source: type[Any], handler: Callable[[Any], CoreSchema]
+    ) -> core_schema.CoreSchema:
+        def f(value, serializer):
+            return serializer(value.timestamp())
+        return core_schema.with_info_after_validator_function(
+                cls._validate, core_schema.float_schema(),
+                serialization=core_schema.wrap_serializer_function_ser_schema(f, return_schema=core_schema.float_schema())
+        )
+
+    @classmethod
+    def _validate(cls, __input_value: Any, _: Any) -> "Epoch":
+        return EPOCH + datetime.timedelta(seconds=__input_value)
+
+

tests/test_epoch.py

@@ -0,0 +1,37 @@
+import datetime
+
+import pytest
+
+from pydantic_extra_types.epoch import Epoch
+
+@pytest.mark.parametrize('type_',[(int,),(float,)], ids=["integer", "float"])
+def test_type(type_):
+    from pydantic import BaseModel
+
+    class A(BaseModel):
+        epoch: Epoch
+
+    now = datetime.datetime.now(tz=datetime.timezone.utc)
+    ts = type_(now.timestamp())
+    a = A.model_validate({"epoch":ts})
+    v = a.model_dump()
+    assert v["epoch"] == ts
+
+    b = A.model_construct(epoch=now)
+
+    v = b.model_dump()
+    assert v["epoch"] == ts
+
+    c = A.model_validate(dict(epoch=ts))
+    v = c.model_dump()
+    assert v["epoch"] == ts
+
+
+def test_schema():
+    from pydantic import BaseModel
+    class A(BaseModel):
+        dt: Epoch
+
+    v = A.model_json_schema()
+    assert (dt:=v["properties"]["dt"])["type"] == "number" and dt["format"] == "date-time"
+