feat: add --resource-name-alias flag to resolve namespace collisions

packages/gapic-generator/gapic/schema/api.py

@@ -1689,6 +1689,7 @@ def _load_message(
                 documentation=self.docs.get(path, self.EMPTY),
             ),
             oneofs=oneofs,
+            resource_name_aliases=self.opts.resource_name_aliases,
         )
         return self.proto_messages[address.proto]
 

packages/gapic-generator/gapic/schema/wrappers.py

@@ -520,6 +520,7 @@ class MessageType:
         default_factory=metadata.Metadata,
     )
     oneofs: Optional[Mapping[str, "Oneof"]] = None
+    resource_name_aliases: Mapping[str, str] = dataclasses.field(default_factory=dict)
 
     def __getattr__(self, name):
         return getattr(self.message_pb, name)
@@ -688,7 +689,13 @@ def resource_path(self) -> Optional[str]:
     @property
     def resource_type(self) -> Optional[str]:
         resource = self.options.Extensions[resource_pb2.resource]
-        return resource.type[resource.type.find("/") + 1 :] if resource else None
+        if not resource:
+            return None
+
+        default_type = resource.type[resource.type.find("/") + 1 :]
+
+        # 2. Safely read from the class's own strictly-typed dictionary!
+        return self.resource_name_aliases.get(resource.type, default_type)
 
     @property
     def resource_type_full_path(self) -> Optional[str]:
@@ -2323,6 +2330,25 @@ def gen_indirect_resources_used(message):
             key=lambda m: m.resource_type_full_path or m.name
         )
 
+        # Fail-fast collision detection
+        seen_types: Dict[str, "MessageType"] = {}
+        for msg in sorted_messages:
+            res_type = msg.resource_type
+            if not res_type:
+                continue
+
+            if res_type in seen_types:
+                incumbent = seen_types[res_type]
+                raise ValueError(
+                    f"\n\nFatal: Namespace collision detected for resource type '{res_type}'.\n"
+                    f"Resources '{incumbent.resource_type_full_path}' and '{msg.resource_type_full_path}' "
+                    f"both flatten to the exact same method name.\n"
+                    f"To protect backward compatibility, explicitly alias one of these using "
+                    f"the `--resource-name-alias` CLI parameter.\n"
+                    f"Example: --resource-name-alias={msg.resource_type_full_path}:CustomName\n"
+                )
+            seen_types[res_type] = msg
+
         return tuple(sorted_messages)
 
     @utils.cached_property

packages/gapic-generator/gapic/utils/options.py

@@ -51,6 +51,7 @@ class Options:
     rest_numeric_enums: bool = False
     proto_plus_deps: Tuple[str, ...] = dataclasses.field(default=("",))
     gapic_version: str = "0.0.0"
+    resource_name_aliases: Dict[str, str] = dataclasses.field(default_factory=dict)
 
     # Class constants
     PYTHON_GAPIC_PREFIX: str = "python-gapic-"
@@ -72,7 +73,11 @@ class Options:
             # proto plus dependencies delineated by '+'
             # For example, 'google.cloud.api.v1+google.cloud.anotherapi.v2'
             "proto-plus-deps",
-            "gapic-version",  # A version string following https://peps.python.org/pep-0440
+            "gapic-version",  # A version string following https://peps.python.org/pep-0440,
+            # Resolves method name collisions by mapping a fully qualified 
+            # resource path to a custom TitleCase alias.
+            # Format: resource.path/Name:AliasName
+            "resource-name-alias",
         )
     )
 
@@ -188,6 +193,37 @@ def tweak_path(p):
         if len(proto_plus_deps):
             proto_plus_deps = tuple(proto_plus_deps[0].split("+"))
 
+        # Parse the resource name aliases dictionary (Format: "path/to/Resource:AliasName")
+        resource_name_aliases = {}
+        raw_aliases = opts.pop("resource-name-alias", [])
+
+        # Normalize: protoc can return a string (1 flag) or list (multiple flags)
+        if not isinstance(raw_aliases, list):
+            raw_aliases = [raw_aliases]
+
+        # Parse explicitly and safely
+        for mapping in raw_aliases:
+            if not mapping or not mapping.strip():
+                continue
+
+            try:
+                # split(":", 1) ensures we only split on the FIRST colon 
+                res_path, alias_name = mapping.split(":", 1)
+
+                clean_path = res_path.strip()
+                clean_alias = alias_name.strip()
+
+                if not clean_path or not clean_alias:
+                    raise ValueError()
+
+                resource_name_aliases[clean_path] = clean_alias
+
+            except ValueError:
+                warnings.warn(
+                    f"Ignored malformed resource-name-alias: '{mapping}'. "
+                    "Expected format is 'resource.path/Name:AliasName'."
+                )
+
         answer = Options(
             name=opts.pop("name", [""]).pop(),
             namespace=tuple(opts.pop("namespace", [])),
@@ -210,6 +246,7 @@ def tweak_path(p):
             rest_numeric_enums=rest_numeric_enums,
             proto_plus_deps=proto_plus_deps,
             gapic_version=opts.pop("gapic-version", ["0.0.0"]).pop(),
+            resource_name_aliases=resource_name_aliases,
         )
 
         # Note: if we ever need to recursively check directories for sample

packages/gapic-generator/tests/unit/generator/test_options.py

@@ -254,3 +254,60 @@ def test_options_proto_plus_deps():
         "google.apps.script.type.slides",
         "google.apps.script.type",
     )
+
+
+def test_options_resource_name_aliases():
+    with mock.patch.object(warnings, "warn") as warn:
+        opts = Options.build(
+            "resource-name-alias=ces.googleapis.com/Tool:CesTool,"
+            "resource-name-alias=workspace.googleapis.com/Tool:WorkspaceTool,"
+            "resource-name-alias=bad_string_without_colon,"
+            "resource-name-alias=:MissingPath"
+        )
+
+        # Verify the dictionary is perfectly formed
+        expected = {
+            "ces.googleapis.com/Tool": "CesTool",
+            "workspace.googleapis.com/Tool": "WorkspaceTool",
+        }
+        assert opts.resource_name_aliases == expected
+
+        # Verify that warnings were safely emitted for
+        # the two malformed aliases
+        assert warn.call_count == 2
+
+
+def test_options_resource_name_aliases():
+    with mock.patch.object(warnings, "warn") as warn:
+        opts = Options.build(
+            "resource-name-alias=ces.googleapis.com/Tool:CesTool,"
+            "resource-name-alias=workspace.googleapis.com/Tool:WorkspaceTool,"
+            "resource-name-alias=bad_string_without_colon,"
+            "resource-name-alias=:MissingPath,"
+            "resource-name-alias=MissingAlias:,"  # <-- Covers 'not clean_alias' (Line 207)
+            "resource-name-alias= ,"              # <-- Covers 'not mapping.strip()' (Line 202)
+        )
+
+        # Verify the dictionary is perfectly formed
+        expected = {
+            "ces.googleapis.com/Tool": "CesTool",
+            "workspace.googleapis.com/Tool": "WorkspaceTool",
+        }
+        assert opts.resource_name_aliases == expected
+
+        # We expect exactly 3 warnings: 
+        # 1. bad_string
+        # 2. :MissingPath 
+        # 3. MissingAlias: 
+        # (The empty ' ' string safely 'continues' without warning, as intended)
+        assert warn.call_count == 3
+
+
+def test_options_resource_name_aliases_single_string():
+    # Because there is only one flag, the parser treats it as a single string 
+    # instead of a list. This forces the `if not isinstance(..., list):` block to execute
+    opts = Options.build("resource-name-alias=single.googleapis.com/Tool:SingleTool")
+
+    assert opts.resource_name_aliases == {
+        "single.googleapis.com/Tool": "SingleTool"
+    }

packages/gapic-generator/tests/unit/schema/wrappers/test_message.py

@@ -473,3 +473,44 @@ def test_extended_operation_request_response_fields():
 
     actual = poll_request.extended_operation_response_fields
     assert actual == expected
+
+
+def test_message_type_resource_type_with_alias():
+    # 1. Cleanly mock the options
+    opts = descriptor_pb2.MessageOptions()
+    opts.Extensions[resource_pb2.resource].type = "ces.googleapis.com/Tool"
+
+    msg_pb = descriptor_pb2.DescriptorProto(name="MyMessage", options=opts)
+
+    # 2. Explicitly instantiate MessageType here instead
+    # of using `make_message` to inject 'resource_name_aliases' field.
+    message = wrappers.MessageType(
+        message_pb=msg_pb,
+        fields={},
+        nested_enums={},
+        nested_messages={},
+        resource_name_aliases={"ces.googleapis.com/Tool": "CesTool"},
+    )
+
+    # 3. Verify it intercepted the default Protobuf logic
+    # and returned the alias
+    assert message.resource_type == "CesTool"
+
+
+def test_message_type_resource_type_without_alias():
+    # 1. Cleanly mock the options
+    opts = descriptor_pb2.MessageOptions()
+    opts.Extensions[resource_pb2.resource].type = "ces.googleapis.com/Tool"
+    msg_pb = descriptor_pb2.DescriptorProto(name="MyMessage", options=opts)
+
+    # 2. Instantiate WITHOUT any aliases matching this resource
+    message = wrappers.MessageType(
+        message_pb=msg_pb,
+        fields={},
+        nested_enums={},
+        nested_messages={},
+        resource_name_aliases={"some.other/Resource": "OtherAlias"},
+    )
+
+    # 3. Verify it falls back to the default type ("Tool")
+    assert message.resource_type == "Tool"

packages/gapic-generator/tests/unit/schema/wrappers/test_service.py

@@ -14,6 +14,7 @@
 
 import collections
 import itertools
+import pytest
 import typing
 
 from google.api import resource_pb2
@@ -693,3 +694,77 @@ def test_extended_operations_lro_detection():
     # because Service objects can't perform the lookup.
     # Instead we kick that can to the API object and make it do the lookup and verification.
     assert lro.operation_service == "CustomOperations"
+
+
+def test_resource_messages_throws_informative_collision_error():
+    # 1. Create options with explicit colliding resource types AND valid patterns
+    opts_1 = descriptor_pb2.MessageOptions()
+    opts_1.Extensions[resource_pb2.resource].type = "ces.googleapis.com/Tool"
+    opts_1.Extensions[resource_pb2.resource].pattern.append("ces/{ces}/tool")
+
+    opts_2 = descriptor_pb2.MessageOptions()
+    opts_2.Extensions[resource_pb2.resource].type = "workspace.googleapis.com/Tool"
+    opts_2.Extensions[resource_pb2.resource].pattern.append("workspaces/{workspace}/tool")
+
+    # 2. Use the test helpers to build the messages
+    msg_1 = make_message("MessageOne", options=opts_1)
+    msg_2 = make_message("MessageTwo", options=opts_2)
+
+    # 3. Build the service with BOTH methods (so the property loops) 
+    # AND visible_resources (so the internal lookups succeed).
+    service = make_service(
+        name="MyService",
+        methods=(
+            make_method("GetToolOne", input_message=msg_1),
+            make_method("GetToolTwo", input_message=msg_2),
+        ),
+        visible_resources={
+            "ces.googleapis.com/Tool": msg_1,
+            "workspace.googleapis.com/Tool": msg_2,
+        }
+    )
+
+    # 4. Assert that asking the service for its resources throws the custom error
+    expected_error_regex = r"(?s)Namespace collision detected.*--resource-name-alias"
+
+    with pytest.raises(ValueError, match=expected_error_regex):
+        _ = service.resource_messages
+
+
+def test_resource_messages_collision_resolved_by_alias():
+    # 1. Colliding types, but valid patterns
+    opts_1 = descriptor_pb2.MessageOptions()
+    opts_1.Extensions[resource_pb2.resource].type = "ces.googleapis.com/Tool"
+    opts_1.Extensions[resource_pb2.resource].pattern.append("ces/{ces}/tool")
+
+    opts_2 = descriptor_pb2.MessageOptions()
+    opts_2.Extensions[resource_pb2.resource].type = "workspace.googleapis.com/Tool"
+    opts_2.Extensions[resource_pb2.resource].pattern.append("workspaces/{workspace}/tool")
+
+    # 2. Build the messages WITH the alias injected into one of them!
+    msg_1 = wrappers.MessageType(
+        message_pb=descriptor_pb2.DescriptorProto(name="MessageOne", options=opts_1), 
+        fields={}, nested_enums={}, nested_messages={},
+        resource_name_aliases={"ces.googleapis.com/Tool": "CesTool"} # Resolved!
+    )
+    msg_2 = wrappers.MessageType(
+        message_pb=descriptor_pb2.DescriptorProto(name="MessageTwo", options=opts_2), 
+        fields={}, nested_enums={}, nested_messages={}
+    )
+
+    # 3. Build the service
+    service = make_service(
+        name="MyService",
+        methods=(
+            make_method("GetToolOne", input_message=msg_1),
+            make_method("GetToolTwo", input_message=msg_2),
+        ),
+        visible_resources={
+            "ces.googleapis.com/Tool": msg_1,
+            "workspace.googleapis.com/Tool": msg_2,
+        }
+    )
+
+    # 4. Assert that calling the property succeeds and returns BOTH resources safely
+    resources = service.resource_messages
+    assert len(resources) == 2