fix: generate body_content_type param when multiple content types share the same schema

Author: shjellvik

end_to_end_tests/functional_tests/generated_code_execution/test_endpoint_bodies.py

@@ -0,0 +1,81 @@
+"""Functional tests for request body handling."""
+
+import pytest
+
+from end_to_end_tests.functional_tests.helpers import (
+    with_generated_client_fixture,
+    with_generated_code_import,
+    with_generated_code_imports,
+)
+
+_DUPLICATE_CONTENT_TYPES_SPEC = """
+components:
+  schemas:
+    MyBody:
+      type: object
+      properties:
+        name:
+          type: string
+      required:
+        - name
+paths:
+  /my-endpoint:
+    post:
+      operationId: my_endpoint
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/MyBody'
+          application/x-www-form-urlencoded:
+            schema:
+              $ref: '#/components/schemas/MyBody'
+          multipart/form-data:
+            schema:
+              $ref: '#/components/schemas/MyBody'
+      responses:
+        '200':
+          description: Success
+"""
+
+
+@with_generated_client_fixture(_DUPLICATE_CONTENT_TYPES_SPEC)
+@with_generated_code_imports(
+    ".models.MyBody",
+)
+@with_generated_code_import(".api.default.my_endpoint._get_kwargs", alias="get_kwargs")
+class TestDuplicateContentTypesUseSameRef:
+    """When all content types in a requestBody reference the same $ref schema,
+    the generated code should use a body_content_type parameter for dispatch
+    instead of isinstance checks (which would all pass for the same type)."""
+
+    def test_defaults_to_json(self, MyBody, get_kwargs):
+        """Without specifying body_content_type, the default content type (first in spec) is used."""
+        body = MyBody(name="test")
+        result = get_kwargs(body=body)
+        assert "json" in result, "Expected JSON body by default"
+        assert result.get("headers", {}).get("Content-Type") == "application/json"
+
+    def test_form_urlencoded(self, MyBody, get_kwargs):
+        """Passing body_content_type='application/x-www-form-urlencoded' sends form data."""
+        body = MyBody(name="test")
+        result = get_kwargs(body=body, body_content_type="application/x-www-form-urlencoded")
+        assert "data" in result, "Expected form-urlencoded body"
+        assert result.get("headers", {}).get("Content-Type") == "application/x-www-form-urlencoded"
+
+    def test_multipart(self, MyBody, get_kwargs):
+        """Passing body_content_type='multipart/form-data' sends multipart data."""
+        body = MyBody(name="test")
+        result = get_kwargs(body=body, body_content_type="multipart/form-data")
+        assert "files" in result, "Expected multipart body"
+
+    def test_json_and_multipart_are_exclusive(self, MyBody, get_kwargs):
+        """JSON and multipart dispatches must be mutually exclusive (not both applied)."""
+        body = MyBody(name="test")
+        json_result = get_kwargs(body=body, body_content_type="application/json")
+        assert "files" not in json_result
+        assert "data" not in json_result
+
+        multipart_result = get_kwargs(body=body, body_content_type="multipart/form-data")
+        assert "json" not in multipart_result
+        assert "data" not in multipart_result

openapi_python_client/parser/openapi.py

@@ -483,14 +483,46 @@ def iter_all_parameters(self) -> Iterator[tuple[oai.ParameterLocation, Property]
         yield from ((oai.ParameterLocation.HEADER, param) for param in self.header_parameters)
         yield from ((oai.ParameterLocation.COOKIE, param) for param in self.cookie_parameters)
 
+    @property
+    def bodies_with_content_type_dispatch(self) -> bool:
+        """True when multiple bodies share the same Python type, making isinstance dispatch non-functional.
+
+        When all content types reference the same $ref schema they resolve to the same Python type.
+        isinstance checks would all pass, causing every branch to execute (last one wins).
+        In this case the generated code uses a ``body_content_type`` string parameter instead.
+        """
+        type_strings = [body.prop.get_type_string() for body in self.bodies]
+        return len(type_strings) != len(set(type_strings))
+
+    @property
+    def unique_body_types(self) -> list[Body]:
+        """Bodies with duplicate Python types removed (first occurrence kept).
+
+        Used for type annotations when bodies_with_content_type_dispatch is True.
+        """
+        seen: set[str] = set()
+        result = []
+        for body in self.bodies:
+            ts = body.prop.get_type_string()
+            if ts not in seen:
+                seen.add(ts)
+                result.append(body)
+        return result
+
     def list_all_parameters(self) -> list[Property]:
         """Return a list of all the parameters of this endpoint"""
+        body_props: list[Property]
+        if self.bodies_with_content_type_dispatch:
+            # Deduplicate: multiple bodies share a type, only emit each unique type once
+            body_props = [body.prop for body in self.unique_body_types]
+        else:
+            body_props = [body.prop for body in self.bodies]
         return (
             self.path_parameters
             + self.query_parameters
             + self.header_parameters
             + self.cookie_parameters
-            + [body.prop for body in self.bodies]
+            + body_props
         )
 
 

openapi_python_client/templates/endpoint_macros.py.jinja

@@ -122,12 +122,21 @@ client: AuthenticatedClient | Client,
 {% if endpoint.bodies | length == 1 %}
 body: {{ endpoint.bodies[0].prop.get_type_string() }}{% if not endpoint.bodies[0].prop.required %} = UNSET{% endif %},
 {% elif endpoint.bodies | length > 1 %}
+{% if endpoint.bodies_with_content_type_dispatch %}
+body:
+    {%- for body in endpoint.unique_body_types -%}{% set body_required = body_required and body.prop.required %}
+    {{ body.prop.get_type_string(no_optional=True) }} {% if not loop.last %} | {% endif %}
+    {%- endfor -%}{% if not body_required %} | Unset = UNSET{% endif %}
+,
+body_content_type: str = "{{ endpoint.bodies[0].content_type }}",
+{% else %}
 body:
     {%- for body in endpoint.bodies -%}{% set body_required = body_required and body.prop.required %}
     {{ body.prop.get_type_string(no_optional=True) }} {% if not loop.last %} | {% endif %}
     {%- endfor -%}{% if not body_required %} | Unset = UNSET{% endif %}
 ,
 {% endif %}
+{% endif %}
 {# query parameters #}
 {% for parameter in endpoint.query_parameters %}
 {{ parameter.to_string() }},
@@ -151,6 +160,9 @@ client=client,
 {% endif %}
 {% if endpoint.bodies | length > 0 %}
 body=body,
+{% if endpoint.bodies_with_content_type_dispatch %}
+body_content_type=body_content_type,
+{% endif %}
 {% endif %}
 {% for parameter in endpoint.query_parameters %}
 {{ parameter.python_name }}={{ parameter.python_name }},

openapi_python_client/templates/endpoint_module.py.jinja

@@ -47,11 +47,19 @@ def _get_kwargs(
     }
 
 {% if endpoint.bodies | length > 1 %}
+{% if endpoint.bodies_with_content_type_dispatch %}
+{% for body in endpoint.bodies %}
+    {{ "if" if loop.first else "elif" }} body_content_type == "{{ body.content_type }}":
+        {{ body_to_kwarg(body) | indent(8) }}
+        headers["Content-Type"] = "{{ body.content_type }}"
+{% endfor %}
+{% else %}
 {% for body in endpoint.bodies %}
     if isinstance(body, {{body.prop.get_type_string(no_optional=True) }}):
         {{ body_to_kwarg(body) | indent(8) }}
         headers["Content-Type"] = "{{ body.content_type }}"
 {% endfor %}
+{% endif %}
 {% elif endpoint.bodies | length == 1 %}
 {% set body = endpoint.bodies[0] %}
     {{ body_to_kwarg(body) | indent(4) }}