diff --git a/.chronus/changes/python-union-etag-regressions-2026-09-18.md b/.chronus/changes/python-union-etag-regressions-2026-09-18.md new file mode 100644 index 00000000000..3c2739a2ccf --- /dev/null +++ b/.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. diff --git a/packages/http-client-python/generator/pygen/codegen/serializers/unions_serializer.py b/packages/http-client-python/generator/pygen/codegen/serializers/unions_serializer.py index b0c7ce09a57..545d08f8e52 100644 --- a/packages/http-client-python/generator/pygen/codegen/serializers/unions_serializer.py +++ b/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, ) diff --git a/packages/http-client-python/generator/pygen/codegen/templates/unions.py.jinja2 b/packages/http-client-python/generator/pygen/codegen/templates/unions.py.jinja2 index 44908503e8c..b2a1bb74a02 100644 --- a/packages/http-client-python/generator/pygen/codegen/templates/unions.py.jinja2 +++ b/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 %} diff --git a/packages/http-client-python/generator/pygen/preprocess/__init__.py b/packages/http-client-python/generator/pygen/preprocess/__init__.py index 1e0a8a82579..4537139def1 100644 --- a/packages/http-client-python/generator/pygen/preprocess/__init__.py +++ b/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") diff --git a/packages/http-client-python/tests/unit/test_typeddict.py b/packages/http-client-python/tests/unit/test_typeddict.py index d0817d6e230..60e3114e810 100644 --- a/packages/http-client-python/tests/unit/test_typeddict.py +++ b/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 ---------- diff --git a/packages/http-client-python/tests/unit/test_typeddict_overloads.py b/packages/http-client-python/tests/unit/test_typeddict_overloads.py index e4298f84083..5686169c1d0 100644 --- a/packages/http-client-python/tests/unit/test_typeddict_overloads.py +++ b/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():