googleapis · viacheslav-rostovtsev · May 11, 2026 · May 11, 2026
@@ -379,7 +379,12 @@ def _render_template(
 
     def _is_desired_transport(self, template_name: str, opts: Options) -> bool:
         """Returns true if template name contains a desired transport"""
-        desired_transports = ["__init__", "base", "README"] + opts.transport
+        desired_transports = [
+            "__init__",
+            "base",
+            "README",
+            "resumable",
+        ] + opts.transport
         return any(transport in template_name for transport in desired_transports)
 
     def _get_file(

@@ -966,6 +966,11 @@ def enforce_valid_library_settings(
         if all_errors:
             raise ClientLibrarySettingsError(yaml.dump(all_errors))
 
+    @cached_property
+    def has_resumable_upload(self) -> bool:
+        """Return True if any service in the API has resumable upload enabled."""
+        return any(s.has_resumable_upload for s in self.services.values())
+
     @cached_property
     def all_method_settings(self) -> Mapping[str, Sequence[client_pb2.MethodSettings]]:
         """Return a map of all `google.api.client.MethodSettings` to be used

@@ -1938,6 +1938,27 @@ def _ref_types(self, recursive: bool) -> Sequence[Union[MessageType, EnumType]]:
         # Done; return the answer.
         return tuple(answer)
 
+    @property
+    def is_resumable_upload(self) -> bool:
+        """Return True if this method is a resumable upload method, False otherwise."""
+        fullMethodName = self.ident.proto
+        if fullMethodName.startswith("google.ads.googleads.v") and fullMethodName.endswith("YouTubeVideoUploadService.CreateYouTubeVideoUpload"):
+            if any(
+                [
+                    self.client_streaming,
+                    self.server_streaming,
+                    self.lro,
+                    self.extended_lro,
+                    self.paged_result_field,
+                ]
+            ):
+                raise ValueError(
+                    f"Method {self.ident.proto} is a resumable upload but "
+                    "also has other features which are mutually exclusive."
+                )
+            return True
+        return False
+
     @property
     def void(self) -> bool:
         """Return True if this method has no return value, False otherwise."""
@@ -2205,6 +2226,11 @@ def host(self) -> str:
             return self.options.Extensions[client_pb2.default_host]
         return ""
 
+    @property
+    def has_resumable_upload(self) -> bool:
+        """Return whether the service has a resumable upload method."""
+        return any(m.is_resumable_upload for m in self.methods.values())
+
     @property
     def version(self) -> str:
         """Return the API version for this service, if specified.

@@ -24,6 +24,9 @@
             {% for field in method.flattened_fields.values() %}
             {{ field.name }}: Optional[{{ field.ident }}] = None,
             {% endfor %}
+            {% if method.is_resumable_upload %}
+            stream: BinaryIO,
+            {% endif %}
             {% else %}
             requests: Optional[Iterator[{{ method.input.ident }}]] = None,
             *,
@@ -60,6 +63,10 @@
                 on the ``request`` instance; if ``request`` is provided, this
                 should not be set.
             {% endfor %}
+            {% if method.is_resumable_upload %}
+            stream (BinaryIO):
+                The stream to upload.
+            {% endif %}
             {% else %}
             requests (Iterator[{{ method.input.ident.sphinx }}]):
                 The request object iterator.{{ " " }}
@@ -140,6 +147,29 @@
             {% endfor %} {# method.flattened_fields.items() #}
             {% endif %} {# method.client_streaming #}
 
+        {% if method.is_resumable_upload %}
+        if stream is None:
+            raise ValueError("stream is required for resumable upload methods")
+
+{{ shared_macros.create_metadata(method) }}
+{{ shared_macros.add_api_version_header_to_metadata(service.version) }}
+{{ shared_macros.auto_populate_uuid4_fields(api, method) }}
+
+        # Validate the universe domain.
+        self._validate_universe_domain()
+
+        # Send the request.
+        response = self._rest_resumable_transport.{{ method.name|snake_case }}_resumable(
+            request=request,
+            stream=stream,
+            retry=retry,
+            timeout=timeout,
+            metadata=metadata,
+        )
+
+        # Done; return the response.
+        return response
+        {% else %}
         # Wrap the RPC method; this adds retry and timeout information,
         # and friendly error handling.
         rpc = self._transport._wrapped_methods[self._transport.{{ method.transport_safe_name|snake_case}}]
@@ -191,6 +221,7 @@
         # Done; return the response.
         return response
         {% endif %}
+        {% endif %}
         {{ "\n" }}
 
 {% endmacro %}

@@ -15,7 +15,7 @@ import json
 import logging as std_logging
 import os
 import re
-from typing import Dict, Callable, Mapping, MutableMapping, MutableSequence, Optional, {% if service.any_server_streaming %}Iterable, {% endif %}{% if service.any_client_streaming %}Iterator, {% endif %}Sequence, Tuple, Type, Union, cast
+from typing import Dict, Callable, Mapping, MutableMapping, MutableSequence, Optional, {% if service.any_server_streaming %}Iterable, {% endif %}{% if service.any_client_streaming %}Iterator, {% endif %}{% if service.has_resumable_upload %}BinaryIO, {% endif %}Sequence, Tuple, Type, Union, cast
 {% if api.all_method_settings.values()|map(attribute="auto_populated_fields", default=[])|list %}
 import uuid
 {% endif %}
@@ -70,6 +70,9 @@ from google.longrunning import operations_pb2 # type: ignore
 {% endif %}
 {% endfilter %}
 from .transports.base import {{ service.name }}Transport, DEFAULT_CLIENT_INFO
+{% if service.has_resumable_upload %}
+from .transports.rest_resumable import {{ service.name }}RestResumableTransport
+{% endif %}
 {% if 'grpc' in opts.transport %}
 from .transports.grpc import {{ service.grpc_transport_name }}
 from .transports.grpc_asyncio import {{ service.grpc_asyncio_transport_name }}
@@ -497,6 +500,12 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
 
     def __init__(self, *,
             credentials: Optional[ga_credentials.Credentials] = None,
+            {% if service.has_resumable_upload %}
+            developer_token: Optional[str] = None,
+            login_customer_id: Optional[str] = None,
+            linked_customer_id: Optional[str] = None,
+            use_cloud_org_for_api_access: bool = False,
+            {% endif %}
             transport: Optional[Union[str, {{ service.name }}Transport, Callable[..., {{ service.name }}Transport]]] = None,
             client_options: Optional[Union[client_options_lib.ClientOptions, dict]] = None,
             client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
@@ -509,6 +518,21 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
                 credentials identify the application to the service; if none
                 are specified, the client will attempt to ascertain the
                 credentials from the environment.
+            {% if service.has_resumable_upload %}
+            {% set resumable_methods = service.methods.values() | selectattr("is_resumable_upload") | map(attribute="name") | map("snake_case") | list %}
+            developer_token (Optional[str]): The developer token to use for the
+                resumable upload transport. This will be used by the resumable upload methods
+                ({{ resumable_methods | join(", ") }}) ONLY!
+            login_customer_id (Optional[str]): The login customer ID to use for
+                the resumable upload transport. This will be used by the resumable upload methods
+                ({{ resumable_methods | join(", ") }}) ONLY!
+            linked_customer_id (Optional[str]): The linked customer ID to use for
+                the resumable upload transport. This will be used by the resumable upload methods
+                ({{ resumable_methods | join(", ") }}) ONLY!
+            use_cloud_org_for_api_access (bool): Whether to use the cloud org for
+                API access for the resumable upload transport. This will be used by the resumable upload methods
+                ({{ resumable_methods | join(", ") }}) ONLY!
+            {% endif %}
             transport (Optional[Union[str,{{ service.name }}Transport,Callable[..., {{ service.name }}Transport]]]):
                 The transport to use, or a Callable that constructs and returns a new transport.
                 If a Callable is given, it will be called with the same set of initialization
@@ -587,16 +611,41 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
         transport_provided = isinstance(transport, {{ service.name }}Transport)
         if transport_provided:
             # transport is a {{ service.name }}Transport instance.
+            {% if service.has_resumable_upload %}
+            if (self._client_options.credentials_file or api_key_value):
+                raise ValueError("When providing a transport instance, "
+                                 "provide its credentials directly.")
+            {% else %}
             if credentials or self._client_options.credentials_file or api_key_value:
                 raise ValueError("When providing a transport instance, "
                                  "provide its credentials directly.")
+            {% endif %}
             if self._client_options.scopes:
                 raise ValueError(
                     "When providing a transport instance, provide its scopes "
                     "directly."
                 )
             self._transport = cast({{ service.name }}Transport, transport)
             self._api_endpoint = self._transport.host
+            {% if service.has_resumable_upload %}
+            resumable_creds = credentials or getattr(self._transport, "_credentials", None)
+
+            if resumable_creds is None:
+                raise ValueError(
+                    "For the correct REST resumable transport initialization, either the credentials should be explicitly provided as a `credentials` parameter, or the transport passed should contain readable `_credentials` property. E.g. passing a GRPC channel as a transport does not satisfy this since the credentials cannot be extracted from it. This requirement is unique to the {{ service.name }}."
+                )
+
+            self._rest_resumable_transport = {{ service.name }}RestResumableTransport(
+                credentials=resumable_creds,
+                credentials_file=None,
+                host=self._transport.host,
+                scopes=self._transport._scopes,
+                developer_token=developer_token,
+                login_customer_id=login_customer_id,
+                linked_customer_id=linked_customer_id,
+                use_cloud_org_for_api_access=use_cloud_org_for_api_access,
+            )
+            {% endif %}
 
         self._api_endpoint = (self._api_endpoint or
             {{ service.client_name }}._get_api_endpoint(
@@ -663,6 +712,27 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
                 always_use_jwt_access=True,
                 api_audience=self._client_options.api_audience,
             )
+            {% if service.has_resumable_upload %}
+
+            # Initialize a separate transport for the rest resumable uploads
+            self._rest_resumable_transport = {{ service.name }}RestResumableTransport(
+                credentials=credentials,
+                credentials_file=self._client_options.credentials_file,
+                host=self._api_endpoint,
+                scopes=self._client_options.scopes,
+                client_cert_source_for_mtls=self._client_cert_source,
+                quota_project_id=self._client_options.quota_project_id,
+                client_info=client_info,
+                always_use_jwt_access=True,
+                api_audience=self._client_options.api_audience,
+                {% if service.has_resumable_upload %}
+                developer_token=developer_token,
+                login_customer_id=login_customer_id,
+                linked_customer_id=linked_customer_id,
+                use_cloud_org_for_api_access=use_cloud_org_for_api_access,
+                {% endif %}
+            )
+            {% endif %}
 
         if "async" not in str(self._transport):
             if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(std_logging.DEBUG):  # pragma: NO COVER