fix: correctly serialize enum path/query params

Author: Anders Brams

openapi_python/generator/normalize.py

@@ -338,8 +338,40 @@ def _schema_map_to_type(
     ), state
 
 
+def _schema_freeform_object_to_type(schema: dict) -> TypeAnnotation:
+    return _nullable(
+        DictAnnotation(NamedAnnotation("str"), AnyAnnotation()),
+        bool(schema.get("nullable")),
+    )
+
+
+def _enum_values_for_schema(state: _TypeState, schema: dict) -> tuple[object, ...]:
+    if "enum" in schema and isinstance(schema["enum"], list):
+        return tuple(schema["enum"])
+
+    ref = schema.get("$ref")
+    if isinstance(ref, str) and ref.startswith("#/components/schemas/"):
+        component = ref.rsplit("/", 1)[-1]
+        component_schema = safe_get(state.components, "schemas", component, type=dict)
+        if component_schema is not None and isinstance(
+            component_schema.get("enum"), list
+        ):
+            return tuple(component_schema["enum"])
+
+    return ()
+
+
+def _with_enum_values(
+    state: _TypeState, schema: dict, annotation: TypeAnnotation
+) -> TypeAnnotation:
+    values = _enum_values_for_schema(state, schema)
+    if not values:
+        return annotation
+    return _union((annotation, LiteralAnnotation(values)))
+
+
 def _schema_object_to_type(
-    state: _TypeState, schema: dict, hint: str
+    state: _TypeState, schema: dict, hint: str, *, allow_enum_values: bool = False
 ) -> tuple[TypeAnnotation, _TypeState]:
     nullable = bool(schema.get("nullable"))
     name = _type_name_from_hint(hint)
@@ -356,8 +388,13 @@ def _schema_object_to_type(
         prop_title = str(safe_get(prop_schema, "title") or prop_name)
         prop_schema = safe_get(props, prop_name, type=dict) or {}
         field_type, state = _schema_to_type(
-            state, prop_schema, f"{name}{_pascal(prop_title)}"
+            state,
+            prop_schema,
+            f"{name}{_pascal(prop_title)}",
+            allow_enum_values=allow_enum_values,
         )
+        if allow_enum_values:
+            field_type = _with_enum_values(state, prop_schema, field_type)
         fields.append(
             FieldDef(
                 name=prop_name,
@@ -385,6 +422,7 @@ def _schema_to_type(
     hint: str,
     *,
     component_name: str | None = None,
+    allow_enum_values: bool = False,
 ) -> tuple[TypeAnnotation, _TypeState]:
     """
     Takes a JSON schema object and returns the corresponding Python type
@@ -425,16 +463,17 @@ def _schema_to_type(
     if schema_type == "array":
         return _schema_array_to_type(state, schema, hint)
 
-    additional_properties = safe_get(schema, "additionalProperties", type=dict)
-    if (
-        schema_type == "object"
-        and "properties" not in schema
-        and additional_properties is not None
-    ):
-        return _schema_map_to_type(state, schema, hint)
+    additional_properties = schema.get("additionalProperties")
+    if schema_type == "object" and "properties" not in schema:
+        if additional_properties is True:
+            return _schema_freeform_object_to_type(schema), state
+        if isinstance(additional_properties, dict):
+            return _schema_map_to_type(state, schema, hint)
 
     if schema_type == "object" or "properties" in schema:
-        return _schema_object_to_type(state, schema, hint)
+        return _schema_object_to_type(
+            state, schema, hint, allow_enum_values=allow_enum_values
+        )
 
     base = _PRIMITIVES.get(str(schema_type), "Any")
     annotation: TypeAnnotation = (
@@ -444,9 +483,11 @@ def _schema_to_type(
 
 
 def _schema_type(
-    state: _TypeState, schema: dict, hint: str
+    state: _TypeState, schema: dict, hint: str, *, allow_enum_values: bool = False
 ) -> tuple[TypeAnnotation, _TypeState]:
-    return _schema_to_type(state, schema or {}, hint)
+    return _schema_to_type(
+        state, schema or {}, hint, allow_enum_values=allow_enum_values
+    )
 
 
 def _path_symbol(path: str) -> str:
@@ -547,6 +588,7 @@ def _bucket_type(
             "required": list(bucket.required),
         },
         hint,
+        allow_enum_values=True,
     )
 
 

openapi_python/generator/render.py

@@ -471,7 +471,7 @@ def _route_aliases(
 def _render_transport(spec: NormalizedSpec, *, protocol_only: bool) -> str:
     return _render_template(
         "transport.py.j2",
-        typing_imports="Protocol" if protocol_only else "TYPE_CHECKING, Protocol",
+        typing_imports=("Protocol" if protocol_only else "TYPE_CHECKING, Protocol"),
         include_default_transport=not protocol_only,
     )
 

openapi_python/generator/runtime_default.py

@@ -1,12 +1,23 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+from enum import Enum
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
     import httpx
 
 
+def _serialize_value(value: object) -> object:
+    if isinstance(value, Enum):
+        return value.value
+    return value
+
+
+def _serialize_mapping(values: Mapping[str, object] | None) -> dict[str, object]:
+    return {key: _serialize_value(value) for key, value in (values or {}).items()}
+
+
 class RuntimeDefaultTransport:
     def __init__(self, *, client: httpx.Client | None = None) -> None:
         if client is None:
@@ -26,11 +37,16 @@ def request(
         headers: Mapping[str, object] | None,
         body: object | None,
     ) -> object:
-        query_dict = {key: str(value) for key, value in (query or {}).items()}
-        header_dict = {key: str(value) for key, value in (headers or {}).items()}
+        path_params = _serialize_mapping(params)
+        query_dict = {
+            key: str(value) for key, value in _serialize_mapping(query).items()
+        }
+        header_dict = {
+            key: str(value) for key, value in _serialize_mapping(headers).items()
+        }
         response = self._client.request(
             method=method.upper(),
-            url=f"{base_url.rstrip('/')}{route.format(**(params or {}))}",
+            url=f"{base_url.rstrip('/')}{route.format(**path_params)}",
             params=query_dict or None,
             headers=header_dict or None,
             json=body,

openapi_python/generator/templates/default_transport.py.j2

@@ -1,3 +1,13 @@
+def _serialize_value(value: object) -> object:
+    if isinstance(value, Enum):
+        return value.value
+    return value
+
+
+def _serialize_mapping(values: Mapping[str, object] | None) -> dict[str, object]:
+    return {key: _serialize_value(value) for key, value in (values or {}).items()}
+
+
 class DefaultTransport:
     def __init__(self, *, client: httpx.Client | None = None) -> None:
         if client is None:
@@ -17,9 +27,10 @@ class DefaultTransport:
         headers: Mapping[str, object] | None,
         body: object | None,
     ) -> object:
-        formatted_route = route.format(**(params or {}))
-        query_dict = {key: str(value) for key, value in (query or {}).items()}
-        header_dict = {key: str(value) for key, value in (headers or {}).items()}
+        path_params = _serialize_mapping(params)
+        formatted_route = route.format(**path_params)
+        query_dict = {key: str(value) for key, value in _serialize_mapping(query).items()}
+        header_dict = {key: str(value) for key, value in _serialize_mapping(headers).items()}
         request_kwargs: dict[str, object] = {"json": body}
         if isinstance(body, Mapping) and any(
             isinstance(value, (bytes, bytearray)) for value in body.values()
@@ -70,9 +81,10 @@ class DefaultAsyncTransport:
         headers: Mapping[str, object] | None,
         body: object | None,
     ) -> object:
-        formatted_route = route.format(**(params or {}))
-        query_dict = {key: str(value) for key, value in (query or {}).items()}
-        header_dict = {key: str(value) for key, value in (headers or {}).items()}
+        path_params = _serialize_mapping(params)
+        formatted_route = route.format(**path_params)
+        query_dict = {key: str(value) for key, value in _serialize_mapping(query).items()}
+        header_dict = {key: str(value) for key, value in _serialize_mapping(headers).items()}
         request_kwargs: dict[str, object] = {"json": body}
         if isinstance(body, Mapping) and any(
             isinstance(value, (bytes, bytearray)) for value in body.values()

openapi_python/generator/templates/transport.py.j2

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+from enum import Enum
 from typing import {{ typing_imports }}
 
 from .types import RouteLiteral as RouteLiteral

tests/contract/enum_path_params/app.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from enum import Enum
+
+from fastapi import FastAPI
+from pydantic import BaseModel
+
+app = FastAPI()
+
+
+class PipelineName(str, Enum):
+    create_sdwan = "create-sdwan"
+    test_pipeline = "test-pipeline"
+    deploy_sdwan_spoke = "deploy-sdwan-spoke"
+    deploy_sdwan_customer_hub = "deploy-sdwan-customer-hub"
+    deploy_cisco_access_switch = "deploy-cisco-access-switch"
+
+
+class PipelineExecution(BaseModel):
+    name: PipelineName
+    id: int
+
+
+@app.get("/api/v1/pipelines/{name}/executions/{id}", response_model=PipelineExecution)
+def get_pipeline_execution(name: PipelineName, id: int) -> PipelineExecution:
+    return PipelineExecution(name=name, id=id)

tests/contract/enum_path_params/generate.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import importlib
+import json
+from pathlib import Path
+from typing import Any, cast
+
+from app import app
+
+from openapi_python.generator import GenerationRequest, generate_client
+
+
+class _Response:
+    content = b"{}"
+
+    def raise_for_status(self) -> None:
+        pass
+
+    def json(self) -> dict[str, object]:
+        return {}
+
+
+class _HttpClient:
+    def __init__(self) -> None:
+        self.url = ""
+
+    def request(self, **kwargs: Any) -> _Response:
+        self.url = str(kwargs["url"])
+        return _Response()
+
+
+def main() -> None:
+    generate_client(
+        GenerationRequest(
+            output_dir=Path(__file__).parent / "generated",
+            spec_json=json.dumps(app.openapi()),
+            overwrite=True,
+        )
+    )
+    types = importlib.import_module("generated.my_client.types")
+    transport_module = importlib.import_module("generated.my_client.transport")
+    http_client = _HttpClient()
+    transport = transport_module.DefaultTransport(client=cast(Any, http_client))
+
+    transport.request(
+        method="get",
+        route="/api/v1/pipelines/{name}/executions/{id}",
+        base_url="http://testserver",
+        params={
+            "name": types.PipelineName.DEPLOY_CISCO_ACCESS_SWITCH,
+            "id": 1,
+        },
+        query=None,
+        headers=None,
+        body=None,
+    )
+    assert (
+        http_client.url
+        == "http://testserver/api/v1/pipelines/deploy-cisco-access-switch/executions/1"
+    )
+
+
+if __name__ == "__main__":
+    main()

tests/contract/enum_path_params/usage_async.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from typing import assert_type
+
+from generated.my_client import AsyncClient
+from generated.my_client.types import PipelineExecution, PipelineName
+
+client = AsyncClient(base_url="http://testserver")
+
+
+async def main() -> None:
+    result = await client.get("/api/v1/pipelines/{name}/executions/{id}")(
+        params={
+            "name": PipelineName.DEPLOY_CISCO_ACCESS_SWITCH.value,
+            "id": 1,
+        }
+    )
+    assert_type(result, PipelineExecution)
+    assert_type(result["name"], PipelineName)
+    assert_type(result["id"], int)

tests/contract/enum_path_params/usage_sync.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from typing import assert_type
+
+from generated.my_client import Client
+from generated.my_client.types import PipelineExecution, PipelineName
+
+client = Client(base_url="http://testserver")
+
+result = client.get("/api/v1/pipelines/{name}/executions/{id}")(
+    params={
+        "name": PipelineName.DEPLOY_CISCO_ACCESS_SWITCH.value,
+        "id": 1,
+    }
+)
+assert_type(result, PipelineExecution)
+assert_type(result["name"], PipelineName)
+assert_type(result["id"], int)

tests/contract/freeform_objects/app.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import FastAPI
+from pydantic import BaseModel
+
+app = FastAPI()
+
+
+class TaskExecutionDto(BaseModel):
+    id: int
+    log: list[dict[str, Any]]
+    metadata: dict[str, object]
+
+
+@app.get("/task-executions/{task_id}", response_model=TaskExecutionDto)
+def get_task_execution(task_id: int) -> TaskExecutionDto:
+    return TaskExecutionDto(
+        id=task_id,
+        log=[{"message": "started", "attempt": 1}],
+        metadata={"trigger": "manual", "dry_run": False},
+    )