feat: add caching to GapicCallable (#666)

* 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 c97a636.

* 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

* use list type for metadata

* add types to docstring

* fixed lint

* convert to list at init time

---------

Co-authored-by: Daniel Sanche <[email protected]>
Co-authored-by: Daniel Sanche <[email protected]>

Author: parthea

google/api_core/gapic_v1/method.py

@@ -42,24 +42,6 @@ class _MethodDefault(enum.Enum):
 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,44 +73,53 @@ def __init__(
     ):
         self._target = target
         self._retry = retry
+        if isinstance(timeout, (int, float)):
+            timeout = TimeToDeadlineTimeout(timeout=timeout)
         self._timeout = timeout
         self._compression = compression
-        self._metadata = metadata
+        self._metadata = list(metadata) if metadata is not None else None
 
     def __call__(
         self, *args, timeout=DEFAULT, retry=DEFAULT, compression=DEFAULT, **kwargs
     ):
         """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(
@@ -202,8 +193,9 @@ def get_topic(name, timeout=None):
 
     Args:
         func (Callable[Any]): The function to wrap. It should accept an
-            optional ``timeout`` argument. If ``metadata`` is not ``None``, it
-            should accept a ``metadata`` argument.
+            optional ``timeout`` (google.api_core.timeout.Timeout) argument.
+            If ``metadata`` is not ``None``, it should accept a ``metadata``
+            (Sequence[Tuple[str, str]]) argument.
         default_retry (Optional[google.api_core.Retry]): The default retry
             strategy. If ``None``, the method will not retry by default.
         default_timeout (Optional[google.api_core.Timeout]): The default

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