fix: freeform objects

Author: Anders Brams

openapi_python/generator/normalize.py

@@ -469,6 +469,8 @@ def _schema_to_type(
             return _schema_freeform_object_to_type(schema), state
         if isinstance(additional_properties, dict):
             return _schema_map_to_type(state, schema, hint)
+        if additional_properties is None:
+            return _schema_freeform_object_to_type(schema), state
 
     if schema_type == "object" or "properties" in schema:
         return _schema_object_to_type(

tests/contract/freeform_objects/app.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Any
 
 from fastapi import FastAPI
@@ -10,14 +11,53 @@
 
 class TaskExecutionDto(BaseModel):
     id: int
+    task_id: int
+    finished_at: datetime | None
     log: list[dict[str, Any]]
-    metadata: dict[str, object]
+    result: int | None
+    output: dict | None
+    should_retry: bool
+    stacktrace: dict | None
 
 
-@app.get("/task-executions/{task_id}", response_model=TaskExecutionDto)
-def get_task_execution(task_id: int) -> TaskExecutionDto:
-    return TaskExecutionDto(
-        id=task_id,
+class TaskDto(BaseModel):
+    id: int
+    state: int
+    queue_id: int
+    n_attempts: int
+    meta: dict | None
+    payload: dict
+    enqueued_at: datetime
+    started_at: datetime | None
+    finished_at: datetime | None
+    pipeline_execution_id: int | None = None
+    pipeline_node_id: int | None = None
+    executions: list[TaskExecutionDto] | None = None
+
+
+@app.get("/tasks", response_model=list[TaskDto])
+def list_tasks() -> list[TaskDto]:
+    execution = TaskExecutionDto(
+        id=1,
+        task_id=1,
+        finished_at=None,
         log=[{"message": "started", "attempt": 1}],
-        metadata={"trigger": "manual", "dry_run": False},
+        result=None,
+        output={"status": "ok"},
+        should_retry=False,
+        stacktrace=None,
     )
+    return [
+        TaskDto(
+            id=1,
+            state=1,
+            queue_id=1,
+            n_attempts=1,
+            meta=None,
+            payload={"operation": "deploy"},
+            enqueued_at=datetime(2026, 1, 1),
+            started_at=None,
+            finished_at=None,
+            executions=[execution],
+        )
+    ]

tests/contract/freeform_objects/generate.py

@@ -10,39 +10,69 @@
 from openapi_python.generator import GenerationRequest, generate_client
 
 
+def _strip_additional_properties(schema: dict) -> None:
+    match schema:
+        case dict():
+            if schema.get("additionalProperties") is True:
+                schema.pop("additionalProperties")
+            for value in schema.values():
+                _strip_additional_properties(value)
+        case list():
+            for item in schema:
+                _strip_additional_properties(item)
+
+
 def main() -> None:
     output_dir = Path(__file__).parent / "generated"
+    spec = app.openapi()
+    _strip_additional_properties(spec)
     generate_client(
         GenerationRequest(
             output_dir=output_dir,
-            spec_json=json.dumps(app.openapi()),
+            spec_json=json.dumps(spec),
             overwrite=True,
         )
     )
 
     source = (output_dir / "my_client" / "types.py").read_text()
     assert "class TaskExecutionDtoLogItem(TypedDict):" not in source
+    assert "class TaskExecutionDtoOutputVariant(TypedDict):" not in source
+    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 "metadata: 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 "executions: NotRequired[list[TaskExecutionDto] | None]" in source
 
     generated_types = importlib.import_module("generated.my_client.types")
     api_b = FastAPI()
 
-    def get_task_execution() -> object:
-        return {}
+    def list_tasks() -> object:
+        return []
 
-    api_b.get(
-        "/task-executions/{task_id}",
-        response_model=generated_types.TaskExecutionDto,
-    )(get_task_execution)
+    api_b.get("/tasks", response_model=list[generated_types.TaskDto])(list_tasks)
 
-    schema = api_b.openapi()["components"]["schemas"]["TaskExecutionDto"]
-    log_items = schema["properties"]["log"]["items"]
-    metadata = schema["properties"]["metadata"]
+    schemas = api_b.openapi()["components"]["schemas"]
+    task_execution = schemas["TaskExecutionDto"]
+    task = schemas["TaskDto"]
+    log_items = task_execution["properties"]["log"]["items"]
+    output = task_execution["properties"]["output"]["anyOf"][0]
+    stacktrace = task_execution["properties"]["stacktrace"]["anyOf"][0]
+    meta = task["properties"]["meta"]["anyOf"][0]
+    payload = task["properties"]["payload"]
     assert log_items["type"] == "object"
     assert log_items["additionalProperties"] is True
-    assert metadata["type"] == "object"
-    assert metadata["additionalProperties"] is True
+    assert output["type"] == "object"
+    assert output["additionalProperties"] is True
+    assert stacktrace["type"] == "object"
+    assert stacktrace["additionalProperties"] is True
+    assert meta["type"] == "object"
+    assert meta["additionalProperties"] is True
+    assert payload["type"] == "object"
+    assert payload["additionalProperties"] is True
 
 
 if __name__ == "__main__":

tests/contract/freeform_objects/usage_async.py

@@ -3,13 +3,21 @@
 from typing import Any, assert_type
 
 from generated.my_client import AsyncClient
-from generated.my_client.types import TaskExecutionDto
+from generated.my_client.types import TaskDto, TaskExecutionDto
 
 client = AsyncClient(base_url="http://testserver")
 
 
 async def main() -> None:
-    result = await client.get("/task-executions/{task_id}")(params={"task_id": 1})
-    assert_type(result, TaskExecutionDto)
-    assert_type(result["log"], list[dict[str, Any]])
-    assert_type(result["metadata"], dict[str, Any])
+    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["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)

tests/contract/freeform_objects/usage_sync.py

@@ -3,11 +3,19 @@
 from typing import Any, assert_type
 
 from generated.my_client import Client
-from generated.my_client.types import TaskExecutionDto
+from generated.my_client.types import TaskDto, TaskExecutionDto
 
 client = Client(base_url="http://testserver")
 
-result = client.get("/task-executions/{task_id}")(params={"task_id": 1})
-assert_type(result, TaskExecutionDto)
-assert_type(result["log"], list[dict[str, Any]])
-assert_type(result["metadata"], dict[str, Any])
+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["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)