feat: add caching to GapicCallable (#527)upstream-main

* feat: optimize _GapicCallable

* cleaned up metadata lines

* chore: avoid type checks in error wrapper

* Revert "chore: avoid type checks in error wrapper"

This reverts commit c97a6365028f3f04d20f26aa1cc0e3131164f53e.

* add default wrapped function

* fixed decorator order

* fixed spacing

* fixed comment typo

* fixed spacing

* fixed spacing

* removed unneeded helpers

* use caching

* improved metadata parsing

* improved docstring

* fixed logic

* added benchmark test

* update threshold

* run benchmark in loop for testing

* use verbose logs

* Revert testing

* used smaller value

* changed threshold

* removed link in comment

2 files changed, 53 insertions, 41 deletions

diff --git a/google/api_core/gapic_v1/method.py b/google/api_core/gapic_v1/method.py
index 0f14ea9..206549e 100644
--- a/google/api_core/gapic_v1/method.py
+++ b/google/api_core/gapic_v1/method.py
@@ -42,24 +42,6 @@ DEFAULT = _MethodDefault._DEFAULT_VALUE
 so the default should be used."""
 
 
-def _is_not_none_or_false(value):
-    return value is not None and value is not False
-
-
-def _apply_decorators(func, decorators):
-    """Apply a list of decorators to a given function.
-
-    ``decorators`` may contain items that are ``None`` or ``False`` which will
-    be ignored.
-    """
-    filtered_decorators = filter(_is_not_none_or_false, reversed(decorators))
-
-    for decorator in filtered_decorators:
-        func = decorator(func)
-
-    return func
-
-
 class _GapicCallable(object):
     """Callable that applies retry, timeout, and metadata logic.
 
@@ -91,6 +73,8 @@ class _GapicCallable(object):
     ):
         self._target = target
         self._retry = retry
+        if isinstance(timeout, (int, float)):
+            timeout = TimeToDeadlineTimeout(timeout=timeout)
         self._timeout = timeout
         self._compression = compression
         self._metadata = metadata
@@ -100,35 +84,42 @@ class _GapicCallable(object):
     ):
         """Invoke the low-level RPC with retry, timeout, compression, and metadata."""
 
-        if retry is DEFAULT:
-            retry = self._retry
-
-        if timeout is DEFAULT:
-            timeout = self._timeout
-
         if compression is DEFAULT:
             compression = self._compression
-
-        if isinstance(timeout, (int, float)):
-            timeout = TimeToDeadlineTimeout(timeout=timeout)
-
-        # Apply all applicable decorators.
-        wrapped_func = _apply_decorators(self._target, [retry, timeout])
+        if compression is not None:
+            kwargs["compression"] = compression
 
         # Add the user agent metadata to the call.
         if self._metadata is not None:
-            metadata = kwargs.get("metadata", [])
-            # Due to the nature of invocation, None should be treated the same
-            # as not specified.
-            if metadata is None:
-                metadata = []
-            metadata = list(metadata)
-            metadata.extend(self._metadata)
-            kwargs["metadata"] = metadata
-        if self._compression is not None:
-            kwargs["compression"] = compression
+            try:
+                # attempt to concatenate default metadata with user-provided metadata
+                kwargs["metadata"] = (*kwargs["metadata"], *self._metadata)
+            except (KeyError, TypeError):
+                # if metadata is not provided, use just the default metadata
+                kwargs["metadata"] = self._metadata
+
+        call = self._build_wrapped_call(timeout, retry)
+        return call(*args, **kwargs)
+
+    @functools.lru_cache(maxsize=4)
+    def _build_wrapped_call(self, timeout, retry):
+        """
+        Build a wrapped callable that applies retry, timeout, and metadata logic.
+        """
+        wrapped_func = self._target
+        if timeout is DEFAULT:
+            timeout = self._timeout
+        elif isinstance(timeout, (int, float)):
+            timeout = TimeToDeadlineTimeout(timeout=timeout)
+        if timeout is not None:
+            wrapped_func = timeout(wrapped_func)
+
+        if retry is DEFAULT:
+            retry = self._retry
+        if retry is not None:
+            wrapped_func = retry(wrapped_func)
 
-        return wrapped_func(*args, **kwargs)
+        return wrapped_func
 
 
 def wrap_method(
diff --git a/tests/unit/gapic/test_method.py b/tests/unit/gapic/test_method.py
index d966f47..370da50 100644
--- a/tests/unit/gapic/test_method.py
+++ b/tests/unit/gapic/test_method.py
@@ -222,3 +222,24 @@ def test_wrap_method_with_call_not_supported():
     with pytest.raises(ValueError) as exc_info:
         google.api_core.gapic_v1.method.wrap_method(method, with_call=True)
     assert "with_call=True is only supported for unary calls" in str(exc_info.value)
+
+
+def test_benchmark_gapic_call():
+    """
+    Ensure the __call__ method performance does not regress from expectations
+
+    __call__ builds a new wrapped function using passed-in timeout and retry, but
+    subsequent calls are cached
+
+    Note: The threshold has been tuned for the CI workers. Test may flake on
+    slower hardware
+    """
+    from google.api_core.gapic_v1.method import _GapicCallable
+    from google.api_core.retry import Retry
+    from timeit import timeit
+
+    gapic_callable = _GapicCallable(
+        lambda *a, **k: 1, retry=Retry(), timeout=1010, compression=False
+    )
+    avg_time = timeit(lambda: gapic_callable(), number=10_000)
+    assert avg_time < 0.4