fix(http-client-python): removing duplicate union aliases and preserve overload parameter types

.chronus/changes/python-union-etag-regressions-2026-09-18.md

@@ -0,0 +1,7 @@
+---
+changeKind: fix
+packages:
+  - "@typespec/http-client-python"
+---
+
+Prevent duplicate named union aliases, and preserve parameter types in generated body overloads when a flattened parameter is filtered out.

packages/http-client-python/generator/pygen/codegen/serializers/unions_serializer.py

@@ -3,7 +3,7 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 # --------------------------------------------------------------------------
-from ..models import CodeModel
+from ..models import CodeModel, CombinedType
 from ..models.imports import FileImport, ImportType
 from ..models.utils import NamespaceType
 from .import_serializer import FileImportSerializer
@@ -18,9 +18,19 @@ def __init__(
     ):
         super().__init__(code_model=code_model, env=env)
 
+    @property
+    def named_unions(self) -> list[CombinedType]:
+        # The same named union can reach codegen as multiple objects, so collapse
+        # by emitted alias name (keeping insertion order) to avoid duplicate aliases.
+        deduped: dict[str, CombinedType] = {}
+        for union in self.code_model.named_unions:
+            if union.name:
+                deduped.setdefault(union.name, union)
+        return list(deduped.values())
+
     def imports(self) -> FileImport:
         file_import = FileImport(self.code_model)
-        if self.code_model.named_unions:
+        if self.named_unions:
             file_import.add_submodule_import(
                 "typing",
                 "TypeAlias",
@@ -31,7 +41,7 @@ def imports(self) -> FileImport:
                 "Union",
                 ImportType.STDLIB,
             )
-        for nu in self.code_model.named_unions:
+        for nu in self.named_unions:
             file_import.merge(
                 nu.imports(
                     serialize_namespace=self.serialize_namespace,
@@ -44,6 +54,7 @@ def serialize(self) -> str:
         template = self.env.get_template("unions.py.jinja2")
         return template.render(
             code_model=self.code_model,
+            named_unions=self.named_unions,
             imports=FileImportSerializer(self.imports()),
             serializer=self,
         )

packages/http-client-python/generator/pygen/codegen/templates/unions.py.jinja2

@@ -4,6 +4,6 @@
 {% endif %}
 
 {{ imports }}
-{% for nu in code_model.named_unions %}
+{% for nu in named_unions %}
 {{nu.name}}: TypeAlias = {{nu.type_definition()}}
 {% endfor %}

packages/http-client-python/generator/pygen/preprocess/__init__.py

@@ -54,13 +54,14 @@ def add_overload(yaml_data: dict[str, Any], body_type: dict[str, Any], for_flatt
     if yaml_data.get("initialOperation"):
         overload["initialOperation"] = yaml_data["initialOperation"]
 
+    # Reattach shared type objects before filtering parameters so positional
+    # alignment with the original operation is preserved.
+    for overload_p, original_p in zip(overload["parameters"], yaml_data["parameters"]):
+        overload_p["type"] = original_p["type"]
     if for_flatten_params:
         overload["bodyParameter"]["flattened"] = True
     else:
         overload["parameters"] = [p for p in overload["parameters"] if not p.get("inFlattenedBody")]
-    # for yaml sync, we need to make sure all of the responses, parameters, and exceptions' types have the same yaml id
-    for overload_p, original_p in zip(overload["parameters"], yaml_data["parameters"]):
-        overload_p["type"] = original_p["type"]
     update_overload_section(overload, yaml_data, "responses")
     update_overload_section(overload, yaml_data, "exceptions")
 

packages/http-client-python/tests/unit/test_typeddict.py

@@ -7,6 +7,7 @@
 """Tests for TypedDict generation, unions generation, and models-mode interactions."""
 
 from jinja2 import PackageLoader, Environment
+import pytest
 
 from pygen.codegen.models import CodeModel, CombinedType, JSONModelType, DPGModelType, build_type
 from pygen.codegen.models.imports import ImportType, FileImport, TypingSection
@@ -561,6 +562,53 @@ def test_unions_serializer_multiple_member_alias():
     )
 
 
+@pytest.mark.parametrize("member_count", [1, 2])
+def test_unions_serializer_deduplicates_named_aliases(member_count: int):
+    """Equivalent single- and multi-member copies produce one alias declaration."""
+    code_model = _make_code_model(models_mode="dpg")
+    voice_model = _make_model(code_model, "GenerateVoiceAgentRequest", model_cls=DPGModelType)
+    text_model = _make_model(code_model, "GenerateTextAgentRequest", model_cls=DPGModelType)
+    members = [voice_model, text_model][:member_count]
+    first = CombinedType(
+        {"type": "combined", "name": "GenerateAgentRequest"},
+        code_model,
+        members,
+    )
+    duplicate = CombinedType(
+        {"type": "combined", "name": "GenerateAgentRequest"},
+        code_model,
+        members.copy(),
+    )
+    code_model.named_unions = [first, duplicate]
+
+    output = UnionsSerializer(code_model=code_model, env=_make_env()).serialize()
+
+    assert output.count("GenerateAgentRequest: TypeAlias =") == 1
+
+
+def test_unions_serializer_collapses_same_name_aliases():
+    """Two unions sharing an alias name emit a single declaration (first wins)."""
+    code_model = _make_code_model(models_mode="dpg")
+    voice_model = _make_model(code_model, "GenerateVoiceAgentRequest", model_cls=DPGModelType)
+    text_model = _make_model(code_model, "GenerateTextAgentRequest", model_cls=DPGModelType)
+    code_model.named_unions = [
+        CombinedType(
+            {"type": "combined", "name": "GenerateAgentRequest"},
+            code_model,
+            [voice_model],
+        ),
+        CombinedType(
+            {"type": "combined", "name": "GenerateAgentRequest"},
+            code_model,
+            [text_model],
+        ),
+    ]
+
+    output = UnionsSerializer(code_model=code_model, env=_make_env()).serialize()
+
+    assert output.count("GenerateAgentRequest: TypeAlias =") == 1
+
+
 # ---------- typed-dict-only ----------
 
 

packages/http-client-python/tests/unit/test_typeddict_overloads.py

@@ -11,7 +11,9 @@
 ``Single overload definition, multiple required``.  The preprocess plugin must
 instead keep the body as a plain single type so no ``@overload`` is emitted.
 """
-from pygen.preprocess import PreProcessPlugin, add_overloads_for_body_param
+import pytest
+
+from pygen.preprocess import PreProcessPlugin, add_overload, add_overloads_for_body_param
 
 
 def _plugin(models_mode: str, generate_typeddict: bool = True) -> PreProcessPlugin:
@@ -97,25 +99,167 @@ def _named_union_operation(member_count: int) -> dict:
     }
 
 
-def test_named_single_member_union_emits_no_overload():
+def _etag_parameters(*, optional: bool) -> tuple[list[dict], dict, dict]:
+    etag_type = {"type": "string"}
+    match_condition_type = {"type": "sdkcore", "name": "MatchConditions"}
+    return (
+        [
+            {
+                "wireName": "If-Match",
+                "clientName": "etag",
+                "location": "header",
+                "optional": optional,
+                "implementation": "Method",
+                "type": etag_type,
+            },
+            {
+                "wireName": "If-None-Match",
+                "clientName": "match_condition",
+                "location": "header",
+                "etagRole": "ifNoneMatch",
+                "optional": optional,
+                "implementation": "Method",
+                "type": match_condition_type,
+            },
+        ],
+        etag_type,
+        match_condition_type,
+    )
+
+
+@pytest.mark.parametrize("optional", [False, True])
+def test_named_single_member_union_emits_no_overload(optional: bool):
     """The implementation annotation carries the alias without an invalid lone overload."""
     yaml_data = _named_union_operation(member_count=1)
     named_union = yaml_data["bodyParameter"]["type"]
+    etag_parameters, etag_type, match_condition_type = _etag_parameters(optional=optional)
+    yaml_data["parameters"].extend(etag_parameters)
 
     add_overloads_for_body_param(yaml_data)
 
     assert yaml_data["overloads"] == []
     assert yaml_data["bodyParameter"]["type"] is named_union
     assert yaml_data["bodyParameter"]["type"]["name"] == "GenerateAgentRequest"
+    assert yaml_data["parameters"][-2]["type"] is etag_type
+    assert yaml_data["parameters"][-1]["type"] is match_condition_type
+    assert all(parameter["optional"] is optional for parameter in yaml_data["parameters"][-2:])
 
 
-def test_named_multiple_member_union_emits_variant_overloads():
+@pytest.mark.parametrize("optional", [False, True])
+def test_named_multiple_member_union_emits_variant_overloads(optional: bool):
     """Multi-member named unions keep one overload per variant."""
     yaml_data = _named_union_operation(member_count=2)
+    named_union = yaml_data["bodyParameter"]["type"]
+    etag_parameters, etag_type, match_condition_type = _etag_parameters(optional=optional)
+    yaml_data["parameters"].extend(etag_parameters)
 
     add_overloads_for_body_param(yaml_data)
 
     assert len(yaml_data["overloads"]) == 2
+    assert yaml_data["bodyParameter"]["type"] is named_union
+    assert all(
+        overload["bodyParameter"]["type"] is member_type
+        for overload, member_type in zip(yaml_data["overloads"], named_union["types"])
+    )
+    for overload in yaml_data["overloads"]:
+        assert overload["parameters"][-2]["type"] is etag_type
+        assert overload["parameters"][-1]["type"] is match_condition_type
+        assert all(parameter["optional"] is optional for parameter in overload["parameters"][-2:])
+
+
+def test_add_overload_preserves_types_after_filtering_flattened_parameters():
+    """Filtering a flattened parameter must not shift later parameter types."""
+    yaml_data = _named_union_operation(member_count=2)
+    flattened_type = {"type": "string", "name": "FlattenedType"}
+    etag_type = {"type": "string", "name": "EtagType"}
+    match_condition_type = {"type": "sdkcore", "name": "MatchConditions"}
+    yaml_data["parameters"].extend(
+        [
+            {
+                "wireName": "flattened",
+                "clientName": "flattened",
+                "location": "body",
+                "optional": True,
+                "implementation": "Method",
+                "inFlattenedBody": True,
+                "type": flattened_type,
+            },
+            {
+                "wireName": "If-Match",
+                "clientName": "etag",
+                "location": "header",
+                "optional": True,
+                "implementation": "Method",
+                "type": etag_type,
+            },
+            {
+                "wireName": "If-None-Match",
+                "clientName": "match_condition",
+                "location": "header",
+                "etagRole": "ifNoneMatch",
+                "optional": True,
+                "implementation": "Method",
+                "type": match_condition_type,
+            },
+        ]
+    )
+
+    overload = add_overload(yaml_data, yaml_data["bodyParameter"]["type"]["types"][0])
+
+    assert [parameter["clientName"] for parameter in overload["parameters"]] == [
+        "content_type",
+        "etag",
+        "match_condition",
+    ]
+    assert overload["parameters"][1]["type"] is etag_type
+    assert overload["parameters"][2]["type"] is match_condition_type
+
+
+@pytest.mark.parametrize("optional", [False, True])
+def test_spread_body_overloads_preserve_etag_parameter_types(optional: bool):
+    """Flattened, JSON, and binary overloads keep the shared ETag pair."""
+    plugin = _plugin("dpg")
+    cross_language_id = "Contoso.TelephonyTransferTargets"
+    original = _dpg_body_parameter("TelephonyTransferTargets", cross_language_id)["type"]
+    spread_body = _json_spread_body_parameter("ReplaceTransferTargetsRequest", cross_language_id)
+    flattened_type = {"type": "list", "elementType": {"type": "string"}}
+    flattened_parameter = {
+        "wireName": "transfer_targets",
+        "clientName": "transfer_targets",
+        "location": "body",
+        "optional": False,
+        "implementation": "Method",
+        "inFlattenedBody": True,
+        "type": flattened_type,
+    }
+    etag_parameters, etag_type, match_condition_type = _etag_parameters(optional=optional)
+    yaml_data = {
+        "name": "replace",
+        "bodyParameter": spread_body,
+        "parameters": [_content_type_param(), flattened_parameter, *etag_parameters],
+        "overloads": [],
+        "responses": [],
+        "exceptions": [],
+    }
+    code_model = {"types": [original, spread_body["type"]]}
+
+    skip_single_body_json = plugin.add_body_param_type(code_model, spread_body)
+    add_overloads_for_body_param(yaml_data, skip_single_body_json=skip_single_body_json)
+
+    assert len(yaml_data["overloads"]) == 3
+    flattened_overloads = [
+        overload for overload in yaml_data["overloads"] if overload["bodyParameter"].get("flattened")
+    ]
+    assert len(flattened_overloads) == 1
+    assert flattened_overloads[0]["parameters"][1]["type"] is flattened_type
+    for overload in yaml_data["overloads"]:
+        parameters_by_name = {parameter["clientName"]: parameter for parameter in overload["parameters"]}
+        assert parameters_by_name["etag"]["type"] is etag_type
+        assert parameters_by_name["match_condition"]["type"] is match_condition_type
+        assert parameters_by_name["etag"]["optional"] is optional
+        assert parameters_by_name["match_condition"]["optional"] is optional
+        if not overload["bodyParameter"].get("flattened"):
+            assert "transfer_targets" not in parameters_by_name
 
 
 def test_typeddict_only_single_body_emits_no_overload():