googleapis · parthea · Sep 11, 2026 · gemini-code-assist · Sep 11, 2026
@@ -1345,7 +1345,8 @@ def _load_children(
             wrapped = loader(
                 child, address=address, path=path + (i,), resources=resources
             )
-            answer[wrapped.name] = wrapped
+            if wrapped is not None:
+                answer[wrapped.name] = wrapped
         return answer
 
     def _get_oneofs(
@@ -1633,6 +1634,9 @@ def _get_methods(
         # Iterate over the methods and collect them into a dictionary.
         answer: Dict[str, wrappers.Method] = collections.OrderedDict()
         for i, meth_pb in enumerate(methods):
+            if self._is_media_upload_proto(meth_pb) and not self.opts.resumable_upload_prefix:
+                continue
+
             retry, timeout = self._get_retry_and_timeout(service_address, meth_pb)
 
             # Create the method wrapper object.
@@ -1651,11 +1655,37 @@ def _get_methods(
                 output=self.api_messages[meth_pb.output_type.lstrip(".")],
                 retry=retry,
                 timeout=timeout,
+                resumable_upload_prefix=self.opts.resumable_upload_prefix,
             )
 
         # Done; return the answer.
         return answer
 
+    def _is_media_upload_proto(
+        self, meth_pb: descriptor_pb2.MethodDescriptorProto
+    ) -> bool:
+        try:
+            if meth_pb.options:
+                http = meth_pb.options.Extensions[annotations_pb2.http]
+                if getattr(http, "media_upload", None) and getattr(
+                    http.media_upload, "enabled", False
+                ):
+                    return True
+                for binding in getattr(http, "additional_bindings", ()):
+                    if getattr(binding, "media_upload", None) and getattr(
+                        binding.media_upload, "enabled", False
+                    ):
+                        return True
+        except Exception:
+            pass
+
+        # TODO(cl/964122389): TEMPORARY - Remove this hardcoded fallback once
+        # the media_upload annotation is published in cl/964122389 and added to gapic-showcase proto.
+        if meth_pb.name == "UploadMedia":
+            return True
+
+        return False
+
     def _load_message(
         self,
         message_pb: descriptor_pb2.DescriptorProto,

@@ -1499,6 +1499,7 @@ class Method:
     meta: metadata.Metadata = dataclasses.field(
         default_factory=metadata.Metadata,
     )
+    resumable_upload_prefix: str = ""
 
     def __getattr__(self, name):
         return getattr(self.method_pb, name)
@@ -1728,6 +1729,32 @@ def http_opt(self) -> Optional[Dict[str, str]]:
         # TODO(yon-mg): enums for http verbs?
         return answer
 
+    @property
+    def is_resumable_upload(self) -> bool:
+        """Return True if this method is a resumable upload method."""
+        if not self.resumable_upload_prefix:
+            return False
+
+        try:
+            if hasattr(self, "options") and self.options:
+                http = self.options.Extensions[annotations_pb2.http]
+                if getattr(http, "media_upload", None) and getattr(http.media_upload, "enabled", False):
+                    return True
+                for binding in getattr(http, "additional_bindings", ()):
+                    if getattr(binding, "media_upload", None) and getattr(binding.media_upload, "enabled", False):
+                        return True
+        except Exception:
+            pass
+
+        # TODO(cl/964122389): TEMPORARY - Remove this hardcoded fallback once
+        # the media_upload annotation is published in cl/964122389 and added to gapic-showcase proto.
+        pb_name = getattr(self.method_pb, "name", "")
+        method_name = getattr(self, "name", "")
+        if pb_name == "UploadMedia" or method_name == "upload_media":
+            return True
+
+        return False
+
     @property
     def path_params(self) -> Sequence[str]:
         """Return the path parameters found in the http annotation path template"""
@@ -2208,6 +2235,11 @@ def has_pagers(self) -> bool:
         """Return whether the service has paged methods."""
         return any(m.paged_result_field for m in self.methods.values())
 
+    @property
+    def has_resumable_upload_methods(self) -> bool:
+        """Return whether the service has resumable upload methods."""
+        return any(m.is_resumable_upload for m in self.methods.values())
+
     @property
     def host(self) -> str:
         """Return the hostname for this service, if specified.

@@ -53,6 +53,17 @@ except ImportError:  # pragma: NO COVER
 {% endmacro %}
 
 {% macro create_metadata(method) %}
+    {% if method.is_resumable_upload %}
+        metadata = () if metadata is None else metadata
+        resumable_metadata = {
+            "x-goog-upload-protocol": "resumable",
+            "x-goog-upload-command": "start",
+        }
+        existing_keys = {k.lower() for k, _ in metadata}
+        metadata = tuple(metadata) + tuple(
+            (k, v) for k, v in resumable_metadata.items() if k not in existing_keys
+        )
+    {% endif %}
     {% if method.explicit_routing %}
         header_params: dict[str, str] = {}
         {% if not method.client_streaming %}
@@ -132,13 +143,13 @@ from google.longrunning import operations_pb2  # type: ignore
 {% endif %}{# import_ns.has_operations_mixin #}
 {% endmacro %}
 
-{% macro http_options_method(rules) %}
+{% macro http_options_method(rules, is_resumable_upload=False, resumable_upload_prefix="resumable/upload") %}
 @staticmethod
 def _get_http_options():
     http_options: List[Dict[str, str]] = [
     {%- for rule in rules %}{
         'method': '{{ rule.method }}',
-        'uri': '{{ rule.uri }}',
+        'uri': '{% if is_resumable_upload %}/{{ resumable_upload_prefix }}{% endif %}{{ rule.uri }}',
         {% if rule.body %}
         'body': '{{ rule.body }}',
         {% endif %}{# rule.body #}

@@ -10,6 +10,7 @@ import pickle
 import warnings
 from typing import Callable, Dict, Optional, Sequence, Tuple, Union
 
+from google.api_core import exceptions as core_exceptions
 from google.api_core import grpc_helpers
 {% if service.has_lro %}
 from google.api_core import operations_v1
@@ -49,6 +50,9 @@ from google.longrunning import operations_pb2 # type: ignore
 {% endif %}
 {% endfilter %}
 from .base import {{ service.name }}Transport, DEFAULT_CLIENT_INFO
+{% if service.has_resumable_upload_methods %}
+from .rest import {{ service.name }}RestTransport
+{% endif %}
 
 try:
     from google.api_core import client_logging  # type: ignore
@@ -353,11 +357,27 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
         # gRPC handles serialization and deserialization, so we just need
         # to pass in the functions for each.
         if '{{ method.transport_safe_name|snake_case }}' not in self._stubs:
+            {% if method.is_resumable_upload %}
+            if not self._credentials:
+                def _error_stub(*args, **kwargs):
+                    raise core_exceptions.GoogleAPICallError(
+                        "Resumable upload methods operate over REST and cannot be invoked when the transport is initialized with a pre-constructed gRPC channel. Please supply credentials directly instead of a gRPC channel to use resumable upload functionality."
+                    )
+                self._stubs['{{ method.transport_safe_name|snake_case }}'] = _error_stub
+            else:
+                rest_transport = {{ service.name }}RestTransport(
+                    host=self._host,
+                    credentials=self._credentials,
+                    client_info=self._client_info,
+                )
+                self._stubs['{{ method.transport_safe_name|snake_case }}'] = rest_transport.{{ method.transport_safe_name|snake_case }}
+            {% else %}
             self._stubs['{{ method.transport_safe_name|snake_case }}'] = self._logged_channel.{{ method.grpc_stub_type }}(
                 '/{{ '.'.join(method.meta.address.package) }}.{{ service.name }}/{{ method.name }}',
                 request_serializer={{ method.input.ident }}.{% if method.input.ident.python_import.module.endswith('_pb2') %}SerializeToString{% else %}serialize{% endif %},
                 response_deserializer={{ method.output.ident }}.{% if method.output.ident.python_import.module.endswith('_pb2') %}FromString{% else %}deserialize{% endif %},
             )
+            {% endif %}
         return self._stubs['{{ method.transport_safe_name|snake_case }}']
     {% endfor %}
 

@@ -54,6 +54,13 @@ from google.longrunning import operations_pb2 # type: ignore
 {% endfilter %}
 from .base import {{ service.name }}Transport, DEFAULT_CLIENT_INFO
 from .grpc import {{ service.name }}GrpcTransport
+{% if service.has_resumable_upload_methods %}
+try:
+    from .rest_asyncio import Async{{ service.name }}RestTransport
+    HAS_ASYNC_REST = True
+except ImportError:
+    HAS_ASYNC_REST = False
+{% endif %}
 
 try:
     from google.api_core import client_logging  # type: ignore
@@ -358,11 +365,31 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
         # gRPC handles serialization and deserialization, so we just need
         # to pass in the functions for each.
         if '{{ method.transport_safe_name|snake_case }}' not in self._stubs:
+            {% if method.is_resumable_upload %}
+            if not self._credentials:
+                async def _error_stub(*args, **kwargs):
+                    raise core_exceptions.GoogleAPICallError(
+                        "Resumable upload methods operate over REST and cannot be invoked when the transport is initialized with a pre-constructed gRPC channel. Please supply credentials directly instead of a gRPC channel to use resumable upload functionality."
+                    )
+                self._stubs['{{ method.transport_safe_name|snake_case }}'] = _error_stub
+            elif HAS_ASYNC_REST:
+                rest_transport = Async{{ service.name }}RestTransport(
+                    host=self._host,
+                    credentials=self._credentials,
+                    client_info=self._client_info,
+                )
+                self._stubs['{{ method.transport_safe_name|snake_case }}'] = rest_transport.{{ method.transport_safe_name|snake_case }}
+            else:
+                async def _unsupported_stub(*args, **kwargs):
+                    raise NotImplementedError("Async REST transport is required for async resumable upload methods.")
+                self._stubs['{{ method.transport_safe_name|snake_case }}'] = _unsupported_stub
+            {% else %}
             self._stubs['{{ method.transport_safe_name|snake_case }}'] = self._logged_channel.{{ method.grpc_stub_type }}(
                 '/{{ '.'.join(method.meta.address.package) }}.{{ service.name }}/{{ method.name }}',
                 request_serializer={{ method.input.ident }}.{% if method.input.ident.python_import.module.endswith('_pb2') %}SerializeToString{% else %}serialize{% endif %},
                 response_deserializer={{ method.output.ident }}.{% if method.output.ident.python_import.module.endswith('_pb2') %}FromString{% else %}deserialize{% endif %},
             )
+            {% endif %}
         return self._stubs['{{ method.transport_safe_name|snake_case }}']
     {% endfor %}
 

@@ -264,7 +264,8 @@ class {{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             pb_resp = resp
             {% endif %}
 
-            json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
+            if response.content and response.content.strip():
+                json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
             {% endif %}{# method.lro #}
             {#- TODO(https://github.com/googleapis/gapic-generator-python/issues/2274): Add debug log before intercepting a request #} 
             resp = self._interceptor.post_{{ method.name|snake_case }}(resp)

@@ -221,7 +221,8 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             pb_resp = resp
             {% endif %}{# if method.output.ident.is_proto_plus_type #}
             content = await response.read()
-            json_format.Parse(content, pb_resp, ignore_unknown_fields=True)
+            if content and content.strip():
+                json_format.Parse(content, pb_resp, ignore_unknown_fields=True)
             {% endif %}{# if method.server_streaming #}
             resp = await self._interceptor.post_{{ method.name|snake_case }}(resp)
             response_metadata = [(k, str(v)) for k, v in response.headers.items()]

@@ -1521,6 +1521,33 @@ def test_{{ service.name|snake_case }}_grpc_asyncio_transport_channel():
     assert transport._ssl_channel_credentials == None
 
 
+{% if service.has_resumable_upload_methods and 'grpc' in opts.transport %}
+{% for method in service.methods.values() if method.is_resumable_upload %}
+def test_{{ service.name|snake_case }}_{{ method.name|snake_case }}_grpc_channel_without_credentials_error():
+    channel = grpc.secure_channel('http://localhost/', grpc.local_channel_credentials())
+    transport = transports.{{ service.name }}GrpcTransport(
+        host="localhost:7469",
+        channel=channel,
+    )
+    with pytest.raises(core_exceptions.GoogleAPICallError) as exc_info:
+        transport.{{ method.transport_safe_name|snake_case }}({{ method.input.ident }}())
+    assert "operate over REST and cannot be invoked when the transport is initialized with a pre-constructed gRPC channel" in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_{{ service.name|snake_case }}_{{ method.name|snake_case }}_grpc_asyncio_channel_without_credentials_error():
+    channel = aio.secure_channel('http://localhost/', grpc.local_channel_credentials())
+    transport = transports.{{ service.name }}GrpcAsyncIOTransport(
+        host="localhost:7469",
+        channel=channel,
+    )
+    with pytest.raises(core_exceptions.GoogleAPICallError) as exc_info:
+        await transport.{{ method.transport_safe_name|snake_case }}({{ method.input.ident }}())
+    assert "operate over REST and cannot be invoked when the transport is initialized with a pre-constructed gRPC channel" in str(exc_info.value)
+{% endfor %}
+{% endif %}
+
+
 # Remove this test when deprecated arguments (api_mtls_endpoint, client_cert_source) are
 # removed from grpc/grpc_asyncio transport constructor.
 @pytest.mark.filterwarnings("ignore::FutureWarning")

@@ -52,6 +52,7 @@ class Options:
     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)
+    resumable_upload_prefix: str = ""
 
     # Class constants
     PYTHON_GAPIC_PREFIX: str = "python-gapic-"
@@ -78,6 +79,8 @@ class Options:
             # resource path to a custom TitleCase alias.
             # Format: resource.path/Name:AliasName
             "resource-name-alias",
+            # Prefix for resumable upload requests
+            "resumable-upload-prefix",
         )
     )
 
@@ -222,6 +225,8 @@ def tweak_path(p):
                     "Expected format is 'resource.path/Name:AliasName'."
                 )
 
+        resumable_upload_prefix = opts.pop("resumable-upload-prefix", [""])[0]
+
         answer = Options(
             name=opts.pop("name", [""]).pop(),
             namespace=tuple(opts.pop("namespace", [])),
@@ -245,6 +250,7 @@ def tweak_path(p):
             proto_plus_deps=proto_plus_deps,
             gapic_version=opts.pop("gapic-version", ["0.0.0"]).pop(),
             resource_name_aliases=resource_name_aliases,
+            resumable_upload_prefix=resumable_upload_prefix,
         )
 
         # Note: if we ever need to recursively check directories for sample