chore: increase async rpc performance

gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2

@@ -326,7 +326,12 @@ class {{ service.async_client_name }}:
             request = {{ method.input.ident }}({% if method.input.ident.package != method.ident.package %}{% for f in method.flattened_fields.values() %}{{ f.name }}={{ f.name }}, {% endfor %}{% endif %})
         {% endif %}{# Cross-package req and flattened fields #}
         {% else %}
-        request = {{ method.input.ident }}(request)
+        # Minor optimization to avoid making a copy if the user passes
+        # in a {{ method.input.ident }}.
+        # There's no risk of modifying the input as we've already verified
+        # there are no flattened fields.
+        if not isinstance(request, {{ method.input.ident }}):
+            request = {{ method.input.ident }}(request)
         {% endif %} {# different request package #}
 
         {# Vanilla python protobuf wrapper types cannot _set_ repeated fields #}
@@ -353,26 +358,9 @@ class {{ service.async_client_name }}:
 
         # Wrap the RPC method; this adds retry and timeout information,
         # and friendly error handling.
-        rpc = gapic_v1.method_async.wrap_method(
-            self._client._transport.{{ method.transport_safe_name|snake_case }},
-            {% if method.retry %}
-            default_retry=retries.AsyncRetry(
-                {% if method.retry.initial_backoff %}initial={{ method.retry.initial_backoff }},{% endif %}
-                {% if method.retry.max_backoff %}maximum={{ method.retry.max_backoff }},{% endif %}
-                {% if method.retry.backoff_multiplier %}multiplier={{ method.retry.backoff_multiplier }},{% endif %}
-                predicate=retries.if_exception_type(
-                    {% for ex in method.retry.retryable_exceptions|sort(attribute="__name__") %}
-                    core_exceptions.{{ ex.__name__ }},
-                    {% endfor %}
-                ),
-                deadline={{ method.timeout }},
-            ),
-            {% endif %}
-            default_timeout={{ method.timeout }},
-            client_info=DEFAULT_CLIENT_INFO,
-        )
-        {% if method.field_headers %}
+        rpc = self._client._transport._wrapped_methods[self._client._transport.{{ method.transport_safe_name|snake_case }}]
 
+        {% if method.field_headers %}
         # Certain fields should be provided within the metadata header;
         # add these here.
         metadata = tuple(metadata) + (

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

@@ -7,6 +7,8 @@ from typing import Awaitable, Callable, Dict, Optional, Sequence, Tuple, Union
 
 from google.api_core import gapic_v1
 from google.api_core import grpc_helpers_async
+from google.api_core import exceptions as core_exceptions
+from google.api_core import retry_async as retries
 {% if service.has_lro %}
 from google.api_core import operations_v1
 {% endif %}
@@ -378,6 +380,32 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
         return self._stubs["test_iam_permissions"]
     {% endif %}
 
+    def _prep_wrapped_messages(self, client_info):
+        # Precompute the wrapped methods.
+        # override base class to use async wrappers
+        self._wrapped_methods = {
+            {% for method in service.methods.values() %}
+            self.{{ method.transport_safe_name|snake_case }}: gapic_v1.method_async.wrap_method(
+                self.{{ method.transport_safe_name|snake_case }},
+                {% if method.retry %}
+                default_retry=retries.AsyncRetry(
+                    {% if method.retry.initial_backoff %}initial={{ method.retry.initial_backoff }},{% endif %}
+                    {% if method.retry.max_backoff %}maximum={{ method.retry.max_backoff }},{% endif %}
+                    {% if method.retry.backoff_multiplier %}multiplier={{ method.retry.backoff_multiplier }},{% endif %}
+                    predicate=retries.if_exception_type(
+                        {% for ex in method.retry.retryable_exceptions|sort(attribute='__name__') %}
+                        core_exceptions.{{ ex.__name__ }},
+                        {% endfor %}
+                    ),
+                    deadline={{ method.timeout }},
+                ),
+                {% endif %}
+                default_timeout={{ method.timeout }},
+                client_info=client_info,
+            ),
+            {% endfor %} {# precomputed wrappers loop #}
+        }
+
     def close(self):
         return self.grpc_channel.close()
 

gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2

@@ -123,6 +123,89 @@ def test_{{ method_name }}_empty_call():
         {% endif %}
 {% endif %}
 
+def test_{{ method_name }}_use_cached_wrapped_rpc():
+    # Clients should use _prep_wrapped_messages to create cached wrapped rpcs,
+    # instead of constructing them on each call
+    with mock.patch("google.api_core.gapic_v1.method.wrap_method") as wrapper_fn:
+        client = {{ service.client_name }}(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport="grpc",
+        )
+
+        # Should wrap all calls on client creation
+        assert wrapper_fn.call_count > 0
+        wrapper_fn.reset_mock()
+
+        # Ensure method has been cached
+        assert client._transport.{{method.transport_safe_name|snake_case}} in client._transport._wrapped_methods
+
+        # Replace cached wrapped function with mock
+        mock_rpc = mock.Mock()
+        client._transport._wrapped_methods[client._transport.{{method.transport_safe_name|snake_case}}] = mock_rpc
+
+        request = {}
+        {% if method.client_streaming %}
+        requests = [request]
+        client.{{ method.safe_name|snake_case }}(iter(requests))
+        {% else %}
+        client.{{ method_name }}(request)
+        {% endif %}
+
+        # Establish that the underlying gRPC stub method was called.
+        assert mock_rpc.call_count == 1
+
+        # Operation methods build a cached wrapper on first rpc call
+        # subsequent calls should use the cached wrapper
+        wrapper_fn.reset_mock()
+        {% if method.client_streaming %}
+        client.{{ method.safe_name|snake_case }}(iter(requests))
+        {% else %}
+        client.{{ method_name }}(request)
+        {% endif %}
+
+        # Establish that a new wrapper was not created for this call
+        assert wrapper_fn.call_count == 0
+        assert mock_rpc.call_count == 2
+
+
+async def test_{{ method_name }}_async_use_cached_wrapped_rpc(transport: str = "grpc_asyncio"):
+    # Clients should use _prep_wrapped_messages to create cached wrapped rpcs,
+    # instead of constructing them on each call
+    with mock.patch("google.api_core.gapic_v1.method_async.wrap_method") as wrapper_fn:
+        client = {{ service.async_client_name }}(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport=transport,
+        )
+
+        # Should wrap all calls on client creation
+        assert wrapper_fn.call_count > 0
+        wrapper_fn.reset_mock()
+
+        # Ensure method has been cached
+        assert client._client._transport.{{method.transport_safe_name|snake_case}} in client._client._transport._wrapped_methods
+
+        # Replace cached wrapped function with mock
+        mock_rpc = mock.AsyncMock()
+        client._client._transport._wrapped_methods[client._client._transport.{{method.transport_safe_name|snake_case}}] = mock_rpc
+
+        request = {}
+        {% if method.client_streaming %}
+        requests = [request]
+        {% endif %}
+        {% if method.client_streaming and method.server_streaming %}
+        await client.{{ method.name|snake_case }}(iter(requests))
+        {% elif method.client_streaming and not method.server_streaming %}
+        await (await client.{{ method.name|snake_case }}(iter(requests)))
+        {% else %}
+        await client.{{ method_name }}(request)
+        {% endif %}
+
+        # Establish that the underlying gRPC stub method was called.
+        assert mock_rpc.call_count == 1
+
+        # Establish that a new wrapper was not created for this call
+        assert wrapper_fn.call_count == 0
+
 {% if not full_extended_lro %}
 @pytest.mark.asyncio
 async def test_{{ method_name }}_async(transport: str = 'grpc_asyncio', request_type={{ method.input.ident }}):
@@ -1069,6 +1152,50 @@ def test_{{ method_name }}_rest(request_type):
     {% endfor %}
     {% endif %}
 
+def test_{{ method_name }}_rest_use_cached_wrapped_rpc():
+    # Clients should use _prep_wrapped_messages to create cached wrapped rpcs,
+    # instead of constructing them on each call
+    with mock.patch("google.api_core.gapic_v1.method.wrap_method") as wrapper_fn:
+        client = {{ service.client_name }}(
+            credentials=ga_credentials.AnonymousCredentials(),
+            transport="rest",
+        )
+
+        # Should wrap all calls on client creation
+        assert wrapper_fn.call_count > 0
+        wrapper_fn.reset_mock()
+
+        # Ensure method has been cached
+        assert client._transport.{{method.transport_safe_name|snake_case}} in client._transport._wrapped_methods
+
+        # Replace cached wrapped function with mock
+        mock_rpc = mock.Mock()
+        client._transport._wrapped_methods[client._transport.{{method.transport_safe_name|snake_case}}] = mock_rpc
+
+        request = {}
+        {% if method.client_streaming %}
+        requests = [request]
+        client.{{ method.safe_name|snake_case }}(iter(requests))
+        {% else %}
+        client.{{ method_name }}(request)
+        {% endif %}
+
+        # Establish that the underlying gRPC stub method was called.
+        assert mock_rpc.call_count == 1
+
+        # Operation methods build a cached wrapper on first rpc call
+        # subsequent calls should use the cached wrapper
+        wrapper_fn.reset_mock()
+        {% if method.client_streaming %}
+        client.{{ method.safe_name|snake_case }}(iter(requests))
+        {% else %}
+        client.{{ method_name }}(request)
+        {% endif %}
+
+        # Establish that a new wrapper was not created for this call
+        assert wrapper_fn.call_count == 0
+        assert mock_rpc.call_count == 2
+
 
     {% if method.input.required_fields %}
 def test_{{ method_name }}_rest_required_fields(request_type={{ method.input.ident }}):