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

Start by comparing the synchronous implementation in packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2 with prep_wrapped_messages_async_method in _shared_macros.j2, then review the related PR #18433. Decide among the proposed consolidation approaches and verify that generated sync, AsyncIO gRPC, REST, and AsyncIO REST transports retain their wrapping and retry behavior without the duplicated templates.

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