Epoch - data type for unix timestamps using pydantic-extra-types (#300)

* 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,20 @@ def generate_type_format_to_class():
 
     type_format_to_class["string"]["byte"] = Base64Str
 
+    type_format_to_class["integer"][None] = int
+
+    try:
+        from pydantic_extra_types import epoch
+
+        type_format_to_class["number"]["date-time"] = epoch.Number
+        type_format_to_class["integer"]["date-time"] = epoch.Integer
+
+    except ImportError:
+        pass
+
 
 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))

docs/source/advanced.rst

@@ -484,3 +484,14 @@ Limiting the concurrency to a certain number of clients:
                 break
             else:
                 done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
+
+Epoch Types
+===========
+
+If installed, pydantic-extra-types is used to provide an epoch data type for integers and float values mapping to datetime.datetime.
+
+A :ref:`Document Plugin <plugin:Document>` can be used to modify a description document to add a format: date-time to the numeric type definition for a posix timestamp.
+
+.. code:: yaml
+        type: integer
+        format: date-time

docs/source/plugin.rst

@@ -128,6 +128,16 @@ Using a Document plugin to modify the parsed description document to state the c
     api = OpenAPI.load_sync("https://try.gitea.io/swagger.v1.json", plugins=[ContentType()])
 
 
+Another example is adding the "format" specifier to an epoch timestamp to have it de-serialized as datetime instead of number/integer.
+
+.. code:: python
+
+    class EpochTimestamp(aiopenapi3.plugin.Document):
+        def parsed(self, ctx):
+            ctx.document["components"]["schemas"]["LogEvent"]["properties"]["timestamp"]["format"] = "date-time"
+            return ctx
+
+
 Message
 =======
 

pyproject.toml

@@ -54,7 +54,9 @@ auth = [
 socks = [
     "httpx-socks",
 ]
-
+types =[
+    "pydantic-extra-types @ git+https://github.com/pydantic/pydantic-extra-types.git@3668b3af8ab378c56342c613672aa9415dab865b",
+]
 [project.scripts]
 aiopenapi3 = "aiopenapi3.cli:main"
 
@@ -118,6 +120,7 @@ tests = [
     "bootstrap-flask",
     "ijson",
     "python-multipart>=0.0.6",
+    "pydantic-extra-types @ git+https://github.com/pydantic/pydantic-extra-types.git@3668b3af8ab378c56342c613672aa9415dab865b"
 ]
 
 [tool.pdm]

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/petstore-expanded.yaml

@@ -131,10 +131,13 @@ components:
         - type: object
           required:
             - id
+            - properties
           properties:
             id:
               type: integer
               format: int64
+            created:
+              type: integer
 
     NewPet:
       type: object

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/plugin_test.py

@@ -1,7 +1,7 @@
-import httpx
-
+import datetime
 from pathlib import Path
 
+import httpx
 import yarl
 
 from aiopenapi3 import FileSystemLoader, OpenAPI
@@ -43,6 +43,7 @@ def parsed(self, ctx):
                     },
                 }
             )
+            ctx.document["components"]["schemas"]["Pet"]["allOf"][1]["properties"]["created"]["format"] = "date-time"
         else:
             raise ValueError(f"unexpected url {ctx.url.path} expecting {self.url}")
 
@@ -57,7 +58,7 @@ def sending(self, ctx):
         return ctx
 
     def received(self, ctx):
-        ctx.received = """[{"id":1,"name":"theanimal", "weight": null}]"""
+        ctx.received = """[{"id":1,"name":"theanimal", "created":4711,"weight": null}]"""
         return ctx
 
     def parsed(self, ctx):
@@ -94,3 +95,4 @@ def test_Plugins(httpx_mock, with_plugin_base):
     assert item.id == 3
     assert item.weight == None  # default does not apply as it it unsed
     assert item.color == "red"  # default does not apply
+    assert item.created == datetime.datetime.fromtimestamp(4711, tz=datetime.timezone.utc)

tests/ref_test.py

@@ -65,23 +65,23 @@ def test_allOf_resolution(petstore_expanded):
         pass
     items = pet.model_fields
 
-    assert sorted(items.keys()) == ["id", "name", "tag"]
+    assert sorted(items.keys()) == ["created", "id", "name", "tag"]
 
     def is_nullable(x):
         # Optional[…] or | None
         return typing.get_origin(x.annotation) == typing.Union and type(None) in typing.get_args(x.annotation)
 
     assert sorted(map(lambda x: x[0], filter(lambda y: is_nullable(y[1]), items.items()))) == sorted(
-        ["tag"]
-    ), ref.schema()
+        ["created", "tag"]
+    ), ref.model_json_schema()
 
     def is_required(x):
         # not assign a default '= Field(default=…)' or '= …'
         return x.default == pydantic_core.PydanticUndefined
 
     assert sorted(map(lambda x: x[0], filter(lambda y: is_required(y[1]), items.items()))) == sorted(
         ["id", "name"]
-    ), ref.schema()
+    ), ref.model_json_schema()
 
     assert items["id"].annotation == int
     assert items["name"].annotation == str

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")