refactor(gapic-generator): deduplicate sync/async prep_wrapped_messages and transport method wrapping

[chalmerlowe]

Problem Summary

In packages/gapic-generator/gapic/templates/, there is code duplication and structural asymmetry between synchronous and asynchronous RPC preparation and method wrapping across the generated transport classes:

1. _prep_wrapped_messages Duplication:

   • packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2 (lines 239–280) defines _prep_wrapped_messages(self, client_info) for synchronous transports.
   • packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/_shared_macros.j2 (lines 322–371) defines prep_wrapped_messages_async_method(api, service) for asynchronous transports.
   • These two blocks are ~50 lines of Jinja template code that are nearly identical. Both iterate over service.methods.values() and api.mixin_api_methods.keys(), mapping every RPC to its retry predicate, default timeout, client info, fully-qualified method name, and streaming flags. The only differences are:
     • self._wrap_method vs self._wrap_async_method
     • retries.Retry(...) vs retries.AsyncRetry(...)

2. _wrap_method vs _wrap_async_method Duplication:

   • Both methods are defined side-by-side in base.py.j2 (lines 174–237).
   • Both methods implement identical compatibility checks and argument filtering across three generations of google-api-core (client_options, kind, method_name, is_streaming), delegating respectively to gapic_v1.method.wrap_method and gapic_v1.method_async.wrap_method.
   • Consider moving these to _compat.py to reduce the footprint in generated code.

3. Architectural Asymmetry (Base Class vs Shared Macros):

   • _wrap_async_method is defined directly on the synchronous base class {{ service.name }}Transport in base.py.j2, even though synchronous transports never call it.
   • Conversely, asynchronous _prep_wrapped_messages is not in base.py.j2—it lives as an external Jinja macro in _shared_macros.j2 that is injected into grpc_asyncio.py.j2 and rest_asyncio.py.j2 to override the base class implementation.
   • This mixing makes the transport hierarchy harder to reason about and causes maintenance drift when retry parameters, metadata, or tracing attributes are updated.

---

Deduplication Options to Consider

Option A: Polymorphic Dispatch in Base Transport (Recommended)

Consolidate _prep_wrapped_messages into a single, unified implementation in base.py.j2 by making transport classes declare their retry class and RPC wrapper callable:

# In base.py.j2 ({{ service.name }}Transport):
_is_async: bool = False
_retry_type = retries.Retry

def _wrap_rpc(self, func, *args, **kwargs):
    # Unified wrap helper parameterized by sync vs async
    wrap_func = gapic_v1.method_async.wrap_method if self._is_async else gapic_v1.method.wrap_method
    supports_tracing = _ASYNC_WRAP_METHOD_SUPPORTS_TRACING if self._is_async else _WRAP_METHOD_SUPPORTS_TRACING
    if supports_tracing:
        kwargs["client_options"] = self._client_options
        if self.kind:
            kwargs["kind"] = self.kind
        return wrap_func(func, *args, **kwargs)
    ...

• Pros:
  • _prep_wrapped_messages only exists once in base.py.j2. All transports (gRPC, AsyncIO gRPC, REST, AsyncIO REST) inherit it naturally.
  • Deletes prep_wrapped_messages_async_method from _shared_macros.j2 entirely (~50 lines removed).
  • Eliminates duplicate template loops for mixin and service methods.
• Cons: Requires setting transport class-level attributes (_is_async, _retry_type) on async transports.

Option B: Parameterized Jinja Macro

Unify synchronous and asynchronous message preparation into a single reusable Jinja macro in _shared_macros.j2:

{% macro prep_wrapped_messages(api, service, is_async=False) %}
def _prep_wrapped_messages(self, client_info):
    self._wrapped_methods = {
        {% for method in service.methods.values() %}
        self.{{ method.transport_safe_name|snake_case }}: self.{{ "_wrap_async_method" if is_async else "_wrap_method" }}(
            ...
            {% if method.retry %}
            default_retry=retries.{{ "AsyncRetry" if is_async else "Retry" }}(...),
            {% endif %}
            ...
        ),
        {% endfor %}
    }
{% endmacro %}

• Pros:
  • Keeps generation purely in templates without modifying Python transport class contracts or inheritance semantics.
  • Single source of truth for method iteration and configuration.
• Cons: Still renders two copies of the method across generated files; Jinja syntax becomes slightly more branchy.

Option C: Dedicated AsyncTransport Base Class

Introduce an explicit _Base{{ service.name }}AsyncTransport class (similar to _Base{{ service.name }}RestTransport):

• {{ service.name }}Transport: Sync base class containing sync _prep_wrapped_messages and _wrap_method.

• _Base{{ service.name }}AsyncTransport: Async base class containing async _prep_wrapped_messages and _wrap_async_method.

• Pros:

  • Clean separation of concerns; async wrappers are no longer baked into sync base classes.
  • Idiomatic object-oriented hierarchy.

• Cons:

  • Larger structural change across generated codebases and public type contracts.
  • Requires updating multiple transport inheritance graphs.

---

Related Discussion

• PR feat(observability): implement universal 4-path OpenTelemetry tracing #18433: feat(observability): [WIP] implement universal 4-path OpenTelemetry tracing
• Follow-up from reviewer feedback regarding code duplication between sync and async transports.