clean up

Author: parthea

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2

@@ -307,21 +307,17 @@ session_kwargs: Dict[str, Any] = (
     if isinstance(timeout, (int, float))
     else {}
 )
-session = {{ client_output_ident }}(
+return {{ client_output_ident }}(
     upload_url=upload_url,
     config=config,
     transport=self._session,
     response_type={{ method.output.ident }},
     start_retry=retry if isinstance(retry, {{ retry_class }}) else None,
+    {% if body_spec %}
+    request_body=body,
+    {% endif %}
     **session_kwargs,
 )
-{% if body_spec %}
-session.upload = functools.partial(session.upload, request_body=body or "")  # type: ignore[method-assign]
-{% if not is_async %}
-session.iter_upload = functools.partial(session.iter_upload, request_body=body or "")  # type: ignore[method-assign]
-{% endif %}
-{% endif %}
-return session
 {%- endmacro %}
 
 

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2

@@ -387,19 +387,6 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
                         return _error()
                 self._stubs['{{ method.transport_safe_name|snake_case }}'] = _ErrorStub()
             elif HAS_ASYNC_REST:
-                # TODO(https://github.com/googleapis/google-cloud-python/issues/16268):
-                # Resolve the credential type mismatch between {{ service.grpc_asyncio_transport_name }}
-                # (where grpc_helpers_async.create_channel and the base transport's
-                # google.auth.default() path use synchronous google.auth.credentials.Credentials)
-                # and Async{{ service.name }}RestTransport (where AsyncAuthorizedSession
-                # requires asynchronous google.auth.aio.credentials.Credentials).
-                # Defer instantiating Async{{ service.name }}RestTransport until the
-                # resumable upload RPC is called so that initializing
-                # {{ service.grpc_asyncio_transport_name }} with synchronous
-                # google.auth.credentials.Credentials (e.g. via ADC or service
-                # account helpers) for standard gRPC calls does not fail in
-                # __init__ (_prep_wrapped_messages) when AsyncAuthorizedSession
-                # validates for async credentials.
                 transport = self
                 class _AsyncRestStub:
                     def __call__(self, *args, **kwargs):

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2

@@ -14,7 +14,6 @@ from google.api_core import rest_helpers
 from google.api_core import rest_streaming
 from google.api_core import gapic_v1
 {% if service.has_resumable_upload_methods %}
-import functools
 import urllib.parse
 from google.api_core import resumable_transfer
 {% endif %}

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2

@@ -22,8 +22,6 @@ from google.auth.aio import credentials as ga_credentials_async  # type: ignore
 from google.api_core import exceptions as core_exceptions
 from google.api_core import gapic_v1
 {% if service.has_resumable_upload_methods %}
-import contextlib
-import functools
 import urllib.parse
 from google.api_core import resumable_transfer
 {% endif %}
@@ -93,46 +91,9 @@ DEFAULT_CLIENT_INFO.protobuf_runtime_version = google.protobuf.__version__
 
 {{ shared_macros.create_interceptor_class(api, service, method, is_async=True) }}
 
-{% if service.has_resumable_upload_methods %}
-@dataclasses.dataclass
-class _AsyncResumableUploadSessionAdapter:
-    """Adapts AsyncAuthorizedSession to the aiohttp-like request context manager interface expected by AsyncResumableUploadSession."""
-
-    _session: AsyncAuthorizedSession
-
-    @contextlib.asynccontextmanager
-    async def request(
-        self,
-        method: str,
-        url: str,
-        *,
-        data: Optional[bytes] = None,
-        headers: Optional[Dict[str, str]] = None,
-        timeout: Any = None,
-        **kwargs,
-    ):
-        response = await self._session.request(
-            method,
-            url,
-            data=data,
-            headers=headers,
-            timeout=getattr(timeout, "total", timeout),
-            **kwargs,
-        )
-        response.status = response.status_code  # type: ignore[attr-defined]
-        try:
-            yield response
-        finally:
-            await response.close()
-{% endif %}
-
 @dataclasses.dataclass
 class Async{{service.name}}RestStub:
-    {% if service.has_resumable_upload_methods %}
-    _session: Union[AsyncAuthorizedSession, _AsyncResumableUploadSessionAdapter]
-    {% else %}
     _session: AsyncAuthorizedSession
-    {% endif %}
     _host: str
     _interceptor: Async{{service.name}}RestInterceptor
 
@@ -380,8 +341,12 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
     @property
     def {{method.transport_safe_name|snake_case}}(self) -> Callable[
             [{{method.input.ident}}],
-            {% if method.is_resumable_upload %}{{method.client_output_async.ident}}{% else %}{{method.output.ident}}{% endif %}]:
-        return self._{{method.name}}({% if method.is_resumable_upload %}_AsyncResumableUploadSessionAdapter(self._session){% else %}self._session{% endif %}, self._host, self._interceptor)  # type: ignore
+            {% if method.is_resumable_upload %}
+            {{method.client_output_async.ident}}]:
+            {% else %}
+            {{method.output.ident}}]:
+            {% endif %}
+        return self._{{method.name}}(self._session, self._host, self._interceptor)  # type: ignore
 
     {% endfor %}
     {% if rest_async_io_enabled %}

packages/gapic-generator/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2

@@ -1587,7 +1587,7 @@ async def test_{{ service.name|snake_case }}_{{ method.name|snake_case }}_grpc_a
         mock_rest_transport_cls.return_value = mock_rest_transport
 
         transport = transports.{{ service.name }}GrpcAsyncIOTransport(
-            credentials=async_anonymous_credentials(),
+            credentials=ga_credentials.AnonymousCredentials(),
         )
         mock_rest_transport_cls.assert_not_called()
 
@@ -1607,11 +1607,40 @@ async def test_{{ service.name|snake_case }}_{{ method.name|snake_case }}_grpc_a
         mock_rest_transport.close.assert_called_once()
 
 
+{% if rest_async_io_enabled %}
+@pytest.mark.asyncio
+async def test_{{ service.name|snake_case }}_{{ method.name|snake_case }}_grpc_asyncio_rest_stub_sync_credentials():
+    if not HAS_ASYNC_REST_EXTRA:
+        pytest.skip()
+    transport = transports.{{ service.name }}GrpcAsyncIOTransport(
+        credentials=ga_credentials.AnonymousCredentials(),
+    )
+    assert transport._rest_transport is None
+    stub = transport.{{ method.transport_safe_name|snake_case }}
+    assert transport._rest_transport is None
+
+    result1 = await stub({{ method.input.ident }}())
+    rest_transport = transport._rest_transport
+    assert rest_transport is not None
+    assert isinstance(result1, {{ method.client_output_async.ident }})
+
+    result2 = await stub({{ method.input.ident }}())
+    assert isinstance(result2, {{ method.client_output_async.ident }})
+    assert transport._rest_transport is rest_transport
+
+    with mock.patch.object(
+        rest_transport, "close", wraps=rest_transport.close
+    ) as mock_rest_close:
+        await transport.close()
+        mock_rest_close.assert_called_once()
+{% endif %}
+
+
 @pytest.mark.asyncio
 async def test_{{ service.name|snake_case }}_{{ method.name|snake_case }}_grpc_asyncio_without_async_rest_error():
     with mock.patch.object(transports.grpc_asyncio, "HAS_ASYNC_REST", False):
         transport = transports.{{ service.name }}GrpcAsyncIOTransport(
-            credentials=async_anonymous_credentials(),
+            credentials=ga_credentials.AnonymousCredentials(),
         )
         with pytest.raises(NotImplementedError) as exc_info:
             await transport.{{ method.transport_safe_name|snake_case }}({{ method.input.ident }}())

packages/gapic-generator/tests/integration/goldens/redis/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py

@@ -2426,67 +2426,67 @@ def operations_client(self) -> AsyncOperationsRestClient:
     @property
     def create_instance(self) -> Callable[
             [cloud_redis.CreateInstanceRequest],
-operations_pb2.Operation]:
+            operations_pb2.Operation]:
         return self._CreateInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def delete_instance(self) -> Callable[
             [cloud_redis.DeleteInstanceRequest],
-operations_pb2.Operation]:
+            operations_pb2.Operation]:
         return self._DeleteInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def export_instance(self) -> Callable[
             [cloud_redis.ExportInstanceRequest],
-operations_pb2.Operation]:
+            operations_pb2.Operation]:
         return self._ExportInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def failover_instance(self) -> Callable[
             [cloud_redis.FailoverInstanceRequest],
-operations_pb2.Operation]:
+            operations_pb2.Operation]:
         return self._FailoverInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def get_instance(self) -> Callable[
             [cloud_redis.GetInstanceRequest],
-cloud_redis.Instance]:
+            cloud_redis.Instance]:
         return self._GetInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def get_instance_auth_string(self) -> Callable[
             [cloud_redis.GetInstanceAuthStringRequest],
-cloud_redis.InstanceAuthString]:
+            cloud_redis.InstanceAuthString]:
         return self._GetInstanceAuthString(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def import_instance(self) -> Callable[
             [cloud_redis.ImportInstanceRequest],
-operations_pb2.Operation]:
+            operations_pb2.Operation]:
         return self._ImportInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def list_instances(self) -> Callable[
             [cloud_redis.ListInstancesRequest],
-cloud_redis.ListInstancesResponse]:
+            cloud_redis.ListInstancesResponse]:
         return self._ListInstances(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def reschedule_maintenance(self) -> Callable[
             [cloud_redis.RescheduleMaintenanceRequest],
-operations_pb2.Operation]:
+            operations_pb2.Operation]:
         return self._RescheduleMaintenance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def update_instance(self) -> Callable[
             [cloud_redis.UpdateInstanceRequest],
-operations_pb2.Operation]:
+            operations_pb2.Operation]:
         return self._UpdateInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def upgrade_instance(self) -> Callable[
             [cloud_redis.UpgradeInstanceRequest],
-operations_pb2.Operation]:
+            operations_pb2.Operation]:
         return self._UpgradeInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property

packages/gapic-generator/tests/integration/goldens/redis_selective/google/cloud/redis_v1/services/cloud_redis/transports/rest_asyncio.py

@@ -1338,31 +1338,31 @@ def operations_client(self) -> AsyncOperationsRestClient:
     @property
     def create_instance(self) -> Callable[
             [cloud_redis.CreateInstanceRequest],
-operations_pb2.Operation]:
+            operations_pb2.Operation]:
         return self._CreateInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def delete_instance(self) -> Callable[
             [cloud_redis.DeleteInstanceRequest],
-operations_pb2.Operation]:
+            operations_pb2.Operation]:
         return self._DeleteInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def get_instance(self) -> Callable[
             [cloud_redis.GetInstanceRequest],
-cloud_redis.Instance]:
+            cloud_redis.Instance]:
         return self._GetInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def list_instances(self) -> Callable[
             [cloud_redis.ListInstancesRequest],
-cloud_redis.ListInstancesResponse]:
+            cloud_redis.ListInstancesResponse]:
         return self._ListInstances(self._session, self._host, self._interceptor)  # type: ignore
 
     @property
     def update_instance(self) -> Callable[
             [cloud_redis.UpdateInstanceRequest],
-operations_pb2.Operation]:
+            operations_pb2.Operation]:
         return self._UpdateInstance(self._session, self._host, self._interceptor)  # type: ignore
 
     @property

packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/transports/grpc_asyncio.py

@@ -365,19 +365,6 @@ async def _error():
                         return _error()
                 self._stubs['upload_media'] = _ErrorStub()
             elif HAS_ASYNC_REST:
-                # TODO(https://github.com/googleapis/google-cloud-python/issues/16268):
-                # Resolve the credential type mismatch between ResumableUploadServiceGrpcAsyncIOTransport
-                # (where grpc_helpers_async.create_channel and the base transport's
-                # google.auth.default() path use synchronous google.auth.credentials.Credentials)
-                # and AsyncResumableUploadServiceRestTransport (where AsyncAuthorizedSession
-                # requires asynchronous google.auth.aio.credentials.Credentials).
-                # Defer instantiating AsyncResumableUploadServiceRestTransport until the
-                # resumable upload RPC is called so that initializing
-                # ResumableUploadServiceGrpcAsyncIOTransport with synchronous
-                # google.auth.credentials.Credentials (e.g. via ADC or service
-                # account helpers) for standard gRPC calls does not fail in
-                # __init__ (_prep_wrapped_messages) when AsyncAuthorizedSession
-                # validates for async credentials.
                 transport = self
                 class _AsyncRestStub:
                     def __call__(self, *args, **kwargs):

packages/gapic-generator/tests/integration/goldens/showcase/google/showcase_v1beta1/services/resumable_upload_service/transports/rest.py

@@ -23,7 +23,6 @@
 from google.api_core import rest_helpers
 from google.api_core import rest_streaming
 from google.api_core import gapic_v1
-import functools
 import urllib.parse
 from google.api_core import resumable_transfer
 from google.showcase_v1beta1._compat import transcode_request
@@ -443,17 +442,15 @@ def __call__(self,
                 if isinstance(timeout, (int, float))
                 else {}
             )
-            session = resumable_transfer.ResumableUploadSession(
+            return resumable_transfer.ResumableUploadSession(
                 upload_url=upload_url,
                 config=config,
                 transport=self._session,
                 response_type=resumable_upload.UploadMediaResponse,
                 start_retry=retry if isinstance(retry, retries.Retry) else None,
+                request_body=body,
                 **session_kwargs,
             )
-            session.upload = functools.partial(session.upload, request_body=body or "")  # type: ignore[method-assign]
-            session.iter_upload = functools.partial(session.iter_upload, request_body=body or "")  # type: ignore[method-assign]
-            return session
 
     @property
     def upload_media(self) -> Callable[