fix: freeform JSON objects normalize to mappings instead of dict[str, Any]

Author: Anders Brams

openapi_python/generator/model.py

@@ -20,6 +20,12 @@ class ListAnnotation:
     item: TypeAnnotation
 
 
+@dataclass(frozen=True)
+class MappingAnnotation:
+    key: TypeAnnotation
+    value: TypeAnnotation
+
+
 @dataclass(frozen=True)
 class LiteralAnnotation:
     values: tuple[object, ...]
@@ -44,6 +50,7 @@ class UnionAnnotation:
     AnyAnnotation
     | DictAnnotation
     | ListAnnotation
+    | MappingAnnotation
     | LiteralAnnotation
     | NamedAnnotation
     | TupleAnnotation

openapi_python/generator/normalize.py

@@ -14,6 +14,7 @@
     FieldDef,
     ListAnnotation,
     LiteralAnnotation,
+    MappingAnnotation,
     NamedAnnotation,
     NormalizedSpec,
     OperationDef,
@@ -340,7 +341,7 @@ def _schema_map_to_type(
 
 def _schema_freeform_object_to_type(schema: dict) -> TypeAnnotation:
     return _nullable(
-        DictAnnotation(NamedAnnotation("str"), AnyAnnotation()),
+        MappingAnnotation(NamedAnnotation("str"), AnyAnnotation()),
         bool(schema.get("nullable")),
     )
 

openapi_python/generator/render.py

@@ -15,6 +15,7 @@
     GeneratedArtifact,
     ListAnnotation,
     LiteralAnnotation,
+    MappingAnnotation,
     NamedAnnotation,
     NormalizedSpec,
     OperationDef,
@@ -34,6 +35,8 @@ def _render_annotation(annotation: TypeAnnotation) -> str:
             return f"dict[{_render_annotation(key)}, {_render_annotation(value)}]"
         case ListAnnotation(item):
             return f"list[{_render_annotation(item)}]"
+        case MappingAnnotation(key, value):
+            return f"Mapping[{_render_annotation(key)}, {_render_annotation(value)}]"
         case LiteralAnnotation(values):
             return f"Literal[{', '.join(repr(value) for value in values)}]"
         case NamedAnnotation(name):
@@ -177,6 +180,10 @@ def _annotation_dependencies(annotation: TypeAnnotation, names: set[str]) -> set
             return _annotation_dependencies(key, names) | _annotation_dependencies(
                 value, names
             )
+        case MappingAnnotation(key, value):
+            return _annotation_dependencies(key, names) | _annotation_dependencies(
+                value, names
+            )
         case ListAnnotation(item):
             return _annotation_dependencies(item, names)
         case NamedAnnotation(name):

openapi_python/generator/templates/types.py.j2

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from enum import Enum
 {% if has_field_descriptions -%}
 from importlib import import_module

tests/contract/freeform_objects/app.py

@@ -35,6 +35,11 @@ class TaskDto(BaseModel):
     executions: list[TaskExecutionDto] | None = None
 
 
+class DeploymentConfigRequest(BaseModel):
+    config: dict
+    franck_submission_id: str | None = None
+
+
 @app.get("/tasks", response_model=list[TaskDto])
 def list_tasks() -> list[TaskDto]:
     execution = TaskExecutionDto(
@@ -61,3 +66,8 @@ def list_tasks() -> list[TaskDto]:
             executions=[execution],
         )
     ]
+
+
+@app.post("/deployment-configs", response_model=DeploymentConfigRequest)
+def create_deployment_config(body: DeploymentConfigRequest) -> DeploymentConfigRequest:
+    return body

tests/contract/freeform_objects/generate.py

@@ -40,11 +40,13 @@ def main() -> None:
     assert "class TaskExecutionDtoStacktraceVariant(TypedDict):" not in source
     assert "class TaskDtoMetaVariant(TypedDict):" not in source
     assert "class TaskDtoPayload(TypedDict):" not in source
-    assert "log: list[dict[str, Any]]" in source
-    assert "output: dict[str, Any] | None" in source
-    assert "stacktrace: dict[str, Any] | None" in source
-    assert "meta: dict[str, Any] | None" in source
-    assert "payload: dict[str, Any]" in source
+    assert "from collections.abc import Mapping" in source
+    assert "log: list[Mapping[str, Any]]" in source
+    assert "output: Mapping[str, Any] | None" in source
+    assert "stacktrace: Mapping[str, Any] | None" in source
+    assert "meta: Mapping[str, Any] | None" in source
+    assert "payload: Mapping[str, Any]" in source
+    assert "config: Mapping[str, Any]" in source
     assert "executions: NotRequired[list[TaskExecutionDto] | None]" in source
 
     generated_types = importlib.import_module("generated.my_client.types")

tests/contract/freeform_objects/usage_async.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
-from typing import Any, assert_type
+from collections.abc import Mapping
+from typing import Any, TypedDict, assert_type
 
 from generated.my_client import AsyncClient
-from generated.my_client.types import TaskDto, TaskExecutionDto
+from generated.my_client.types import DeploymentConfigRequest, TaskDto, TaskExecutionDto
 
 client = AsyncClient(base_url="http://testserver")
 
@@ -12,12 +13,30 @@ async def main() -> None:
     result = await client.get("/tasks")()
     assert_type(result, list[TaskDto])
     task = result[0]
-    assert_type(task["meta"], dict[str, Any] | None)
-    assert_type(task["payload"], dict[str, Any])
+    assert_type(task["meta"], Mapping[str, Any] | None)
+    assert_type(task["payload"], Mapping[str, Any])
     assert_type(task["executions"], list[TaskExecutionDto] | None)
     execution = task["executions"][0] if task["executions"] is not None else None
     assert_type(execution, TaskExecutionDto | None)
     if execution is not None:
-        assert_type(execution["log"], list[dict[str, Any]])
-        assert_type(execution["output"], dict[str, Any] | None)
-        assert_type(execution["stacktrace"], dict[str, Any] | None)
+        assert_type(execution["log"], list[Mapping[str, Any]])
+        assert_type(execution["output"], Mapping[str, Any] | None)
+        assert_type(execution["stacktrace"], Mapping[str, Any] | None)
+
+
+class FortigateSDWANSpokeConfig(TypedDict):
+    hostname: str
+    serial_number: str
+
+
+async def create_deployment_config() -> None:
+    config: FortigateSDWANSpokeConfig = {
+        "hostname": "fw01",
+        "serial_number": "FGT123",
+    }
+    body: DeploymentConfigRequest = {
+        "config": config,
+        "franck_submission_id": None,
+    }
+    created = await client.post("/deployment-configs")(body=body)
+    assert_type(created, DeploymentConfigRequest)

tests/contract/freeform_objects/usage_sync.py

@@ -1,21 +1,39 @@
 from __future__ import annotations
 
-from typing import Any, assert_type
+from collections.abc import Mapping
+from typing import Any, TypedDict, assert_type
 
 from generated.my_client import Client
-from generated.my_client.types import TaskDto, TaskExecutionDto
+from generated.my_client.types import DeploymentConfigRequest, TaskDto, TaskExecutionDto
 
 client = Client(base_url="http://testserver")
 
 result = client.get("/tasks")()
 assert_type(result, list[TaskDto])
 task = result[0]
-assert_type(task["meta"], dict[str, Any] | None)
-assert_type(task["payload"], dict[str, Any])
+assert_type(task["meta"], Mapping[str, Any] | None)
+assert_type(task["payload"], Mapping[str, Any])
 assert_type(task["executions"], list[TaskExecutionDto] | None)
 execution = task["executions"][0] if task["executions"] is not None else None
 assert_type(execution, TaskExecutionDto | None)
 if execution is not None:
-    assert_type(execution["log"], list[dict[str, Any]])
-    assert_type(execution["output"], dict[str, Any] | None)
-    assert_type(execution["stacktrace"], dict[str, Any] | None)
+    assert_type(execution["log"], list[Mapping[str, Any]])
+    assert_type(execution["output"], Mapping[str, Any] | None)
+    assert_type(execution["stacktrace"], Mapping[str, Any] | None)
+
+
+class FortigateSDWANSpokeConfig(TypedDict):
+    hostname: str
+    serial_number: str
+
+
+config: FortigateSDWANSpokeConfig = {
+    "hostname": "fw01",
+    "serial_number": "FGT123",
+}
+body: DeploymentConfigRequest = {
+    "config": config,
+    "franck_submission_id": None,
+}
+created = client.post("/deployment-configs")(body=body)
+assert_type(created, DeploymentConfigRequest)

tests/contract/union/usage_async.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from typing import Any, assert_type
 
 from generated.my_client import AsyncClient
@@ -20,8 +21,8 @@ async def use_async_client() -> None:
 
     int_result = await async_client.get("/lookup/{value}")(params=int_params)
     assert_type(int_result, LookupResult)
-    assert_type(int_result["value"], int | str | dict[str, Any])
+    assert_type(int_result["value"], int | str | Mapping[str, Any])
 
     str_result = await async_client.get("/lookup/{value}")(params=str_params)
     assert_type(str_result, LookupResult)
-    assert_type(str_result["value"], int | str | dict[str, Any])
+    assert_type(str_result["value"], int | str | Mapping[str, Any])

tests/contract/union/usage_sync.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
 from typing import Any, assert_type
 
 from generated.my_client import Client
@@ -19,8 +20,8 @@
 
 int_result = client.get("/lookup/{value}")(params=int_params)
 assert_type(int_result, LookupResult)
-assert_type(int_result["value"], int | str | dict[str, Any])
+assert_type(int_result["value"], int | str | Mapping[str, Any])
 
 str_result = client.get("/lookup/{value}")(params=str_params)
 assert_type(str_result, LookupResult)
-assert_type(str_result["value"], int | str | dict[str, Any])
+assert_type(str_result["value"], int | str | Mapping[str, Any])