v20/v3x - epoch datetime values

have a datatype for epoch datetimes
type:integer/number with format:date-time is datetime

c.f. pydantic/pydantic-extra-types#240

Author: commonism

aiopenapi3/model.py

@@ -24,7 +24,7 @@
     from .base import DiscriminatorBase
     from ._types import SchemaType, ReferenceType, PrimitiveTypes, DiscriminatorType
 
-type_format_to_class: dict[str, dict[str, type]] = collections.defaultdict(dict)
+type_format_to_class: dict[str, dict[Optional[str], type]] = collections.defaultdict(dict)
 
 log = logging.getLogger("aiopenapi3.model")
 
@@ -58,11 +58,17 @@ def generate_type_format_to_class():
 
     type_format_to_class["string"]["byte"] = Base64Str
 
+    from aiopenapi3.models import epoch
+
+    type_format_to_class["number"]["date-time"] = epoch.Number
+
+    if "integer" not in type_format_to_class:
+        type_format_to_class["integer"][None] = int
+    type_format_to_class["integer"]["date-time"] = epoch.Integer
+
 
 def class_from_schema(s, _type):
-    if _type == "integer":
-        return int
-    elif _type == "boolean":
+    if _type == "boolean":
         return bool
     a = type_format_to_class[_type]
     b = a.get(s.format, a[None])

aiopenapi3/models/epoch.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import datetime
+from typing import Any, Callable
+
+import pydantic_core.core_schema
+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 _Base(datetime.datetime):
+    TYPE: str = ""
+    SCHEMA: pydantic_core.core_schema.CoreSchema
+
+    @classmethod
+    def __get_pydantic_json_schema__(
+        cls, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler
+    ) -> JsonSchemaValue:
+        field_schema: dict[str, Any] = {}
+        field_schema.update(type=cls.TYPE, format="date-time")
+        return field_schema
+
+    @classmethod
+    def __get_pydantic_core_schema__(
+        cls, source: type[Any], handler: Callable[[Any], CoreSchema]
+    ) -> core_schema.CoreSchema:
+        return core_schema.with_info_after_validator_function(
+            cls._validate,
+            cls.SCHEMA,
+            serialization=core_schema.wrap_serializer_function_ser_schema(cls._f, return_schema=cls.SCHEMA),
+        )
+
+    @classmethod
+    def _validate(cls, __input_value: Any, _: Any) -> datetime.datetime:
+        return EPOCH + datetime.timedelta(seconds=__input_value)
+
+    @classmethod
+    def _f(cls, value: Any, serializer: Callable[[Any], Any]) -> Any:  # pragma: no cover
+        raise NotImplementedError(cls)
+
+
+class Number(_Base):
+    TYPE = "number"
+    SCHEMA = core_schema.float_schema()
+
+    @classmethod
+    def _f(cls, value: Any, serializer: Callable[[float], float]) -> float:
+        ts = value.timestamp()
+        return serializer(ts)
+
+
+class Integer(_Base):
+    TYPE = "integer"
+    SCHEMA = core_schema.int_schema()
+
+    @classmethod
+    def _f(cls, value: Any, serializer: Callable[[int], int]) -> int:
+        ts = value.timestamp()
+        return serializer(int(ts))

tests/conftest.py

@@ -440,6 +440,11 @@ def with_schema_additionalProperties_and_named_properties():
     yield _get_parsed_yaml("schema-additionalProperties-and-named-properties" ".yaml")
 
 
+@pytest.fixture
+def with_schema_date_types():
+    yield _get_parsed_yaml("schema-date-types.yaml")
+
+
 @pytest.fixture
 def with_schema_boolean_v20():
     yield _get_parsed_yaml("schema-boolean-v20.yaml")

tests/fixtures/schema-date-types.yaml

@@ -0,0 +1,18 @@
+openapi: "3.1.0"
+info:
+  version: 1.0.0
+  title: date-time
+
+components:
+  schemas:
+    Integer:
+      type: integer
+      format: date-time
+
+    Number:
+      type: number
+      format: date-time
+
+    String:
+      type: string
+      format: date-time

tests/schema_test.py

@@ -1,6 +1,7 @@
 import copy
 import typing
 import uuid
+from datetime import datetime
 from unittest.mock import MagicMock, patch
 
 from pydantic.fields import FieldInfo
@@ -766,3 +767,26 @@ def test_schema_boolean_v20(with_schema_boolean_v20):
 
     with pytest.raises(ValidationError):
         B.model_validate({"b": 1})
+
+
+def test_schema_date_types(with_schema_date_types):
+    api = OpenAPI("/", with_schema_date_types)
+    Integer = api.components.schemas["Integer"].get_type()
+    Number = api.components.schemas["Number"].get_type()
+    String = api.components.schemas["String"].get_type()
+
+    from datetime import timezone
+
+    now = datetime.now(tz=timezone.utc)
+    ts = now.timestamp()
+    v = Integer.model_validate(c := int(ts))
+    assert isinstance(v.root, datetime)
+    assert v.model_dump() == c
+
+    v = Number.model_validate(ts)
+    assert isinstance(v.root, datetime)
+    assert v.model_dump() == ts
+
+    v = String.model_validate(str(ts))
+    assert isinstance(v.root, datetime)
+    assert v.model_dump_json()[1:-1] == now.strftime("%Y-%m-%dT%H:%M:%S.%fZ")