chore: refactor rest interceptor class into jinja macro

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

@@ -279,3 +279,99 @@ def _wrap_method(self, func, *args, **kwargs):
         kwargs["kind"] = self.kind
     return gapic_v1.method_async.wrap_method(func, *args, **kwargs)
 {% endmacro %}
+
+{# `create_interceptor_class` generates an Interceptor class for
+ # synchronous and asynchronous rest transports
+#}
+{% macro create_interceptor_class(api, service, method, is_async=False) %}
+{% set async_prefix = "async " if is_async else "" %}
+{% set async_method_name_prefix = "Async" if is_async else "" %}
+{% set async_docstring = "Asynchronous " if is_async else "" %}
+{% set async_suffix = "_async" if is_async else "" %}
+
+class {{ async_method_name_prefix }}{{ service.name }}RestInterceptor:
+    """{{ async_docstring }}Interceptor for {{ service.name }}.
+
+    Interceptors are used to manipulate requests, request metadata, and responses
+    in arbitrary ways.
+    Example use cases include:
+    * Logging
+    * Verifying requests according to service or custom semantics
+    * Stripping extraneous information from responses
+
+    These use cases and more can be enabled by injecting an
+    instance of a custom subclass when constructing the {{ async_method_name_prefix }}{{ service.name }}RestTransport.
+
+    .. code-block:: python
+        class MyCustom{{ service.name }}Interceptor({{ service.name }}RestInterceptor):
+    {% for _, method in service.methods|dictsort if not method.client_streaming %}
+            {{ async_prefix }}def pre_{{ method.name|snake_case }}(self, request, metadata):
+                logging.log(f"Received request: {request}")
+                return request, metadata
+
+    {% if not method.void %}
+            {{ async_prefix }}def post_{{ method.name|snake_case }}(self, response):
+                logging.log(f"Received response: {response}")
+                return response
+    {% endif %}
+
+    {% endfor %}
+        transport = {{ async_method_name_prefix }}{{ service.name }}RestTransport(interceptor=MyCustom{{ service.name }}Interceptor())
+        client = {{ async_prefix }}{{ service.client_name }}(transport=transport)
+
+
+    """
+    {% for method in service.methods.values()|sort(attribute="name") if not method.client_streaming and method.http_options %}
+    {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2147): Remove the condition below once async rest transport supports the guarded methods. #}
+    {% if (not is_async) or (is_async and not method.lro and not method.extended_lro and not method.paged_result_field) %}
+    {{ async_prefix }}def pre_{{ method.name|snake_case }}(self, request: {{method.input.ident}}, metadata: Sequence[Tuple[str, str]]) -> Tuple[{{method.input.ident}}, Sequence[Tuple[str, str]]]:
+        """Pre-rpc interceptor for {{ method.name|snake_case }}
+
+        Override in a subclass to manipulate the request or metadata
+        before they are sent to the {{ service.name }} server.
+        """
+        return request, metadata
+
+    {% if not method.void %}
+    {% if not method.server_streaming %}
+    {{ async_prefix }}def post_{{ method.name|snake_case }}(self, response: {{method.output.ident}}) -> {{method.output.ident}}:
+    {% else %}
+    {{ async_prefix }}def post_{{ method.name|snake_case }}(self, response: rest_streaming{{ async_suffix }}.{{ async_method_name_prefix }}ResponseIterator) -> rest_streaming{{ async_suffix }}.{{ async_method_name_prefix }}ResponseIterator:
+    {% endif %}
+        """Post-rpc interceptor for {{ method.name|snake_case }}
+
+        Override in a subclass to manipulate the response
+        after it is returned by the {{ service.name }} server but before
+        it is returned to user code.
+        """
+        return response
+    {% endif %}
+    {% endif %}{# if (not is_async) or (is_async and not method.lro and not method.extended_lro and not method.paged_result_field) #}
+    {% endfor %}
+
+    {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2148): Remove the condition below once mixins are supported for async rest transport. #}
+    {% if not is_async %}
+    {% for name, signature in api.mixin_api_signatures.items() %}
+    {{ async_prefix }}def pre_{{ name|snake_case }}(
+        self, request: {{signature.request_type}}, metadata: Sequence[Tuple[str, str]]
+    ) -> Tuple[{{signature.request_type}}, Sequence[Tuple[str, str]]]:
+        """Pre-rpc interceptor for {{ name|snake_case }}
+
+        Override in a subclass to manipulate the request or metadata
+        before they are sent to the {{ service.name }} server.
+        """
+        return request, metadata
+
+    {{ async_prefix }}def post_{{ name|snake_case }}(
+        self, response: {{signature.response_type}}
+    ) -> {{signature.response_type}}:
+        """Post-rpc interceptor for {{ name|snake_case }}
+
+        Override in a subclass to manipulate the response
+        after it is returned by the {{ service.name }} server but before
+        it is returned to user code.
+        """
+        return response
+    {% endfor %}
+    {% endif %}
+{% endmacro %}

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

@@ -48,85 +48,7 @@ DEFAULT_CLIENT_INFO = gapic_v1.client_info.ClientInfo(
 )
 
 
-class {{ service.name }}RestInterceptor:
-    """Interceptor for {{ service.name }}.
-
-    Interceptors are used to manipulate requests, request metadata, and responses
-    in arbitrary ways.
-    Example use cases include:
-    * Logging
-    * Verifying requests according to service or custom semantics
-    * Stripping extraneous information from responses
-
-    These use cases and more can be enabled by injecting an
-    instance of a custom subclass when constructing the {{ service.name }}RestTransport.
-
-    .. code-block:: python
-        class MyCustom{{ service.name }}Interceptor({{ service.name }}RestInterceptor):
-    {% for _, method in service.methods|dictsort if not method.client_streaming %}
-            def pre_{{ method.name|snake_case }}(self, request, metadata):
-                logging.log(f"Received request: {request}")
-                return request, metadata
-
-    {% if not method.void %}
-            def post_{{ method.name|snake_case }}(self, response):
-                logging.log(f"Received response: {response}")
-                return response
-    {% endif %}
-
-{% endfor %}
-        transport = {{ service.name }}RestTransport(interceptor=MyCustom{{ service.name }}Interceptor())
-        client = {{ service.client_name }}(transport=transport)
-
-
-    """
-    {% for method in service.methods.values()|sort(attribute="name") if not method.client_streaming and method.http_options %}
-    def pre_{{ method.name|snake_case }}(self, request: {{method.input.ident}}, metadata: Sequence[Tuple[str, str]]) -> Tuple[{{method.input.ident}}, Sequence[Tuple[str, str]]]:
-        """Pre-rpc interceptor for {{ method.name|snake_case }}
-
-        Override in a subclass to manipulate the request or metadata
-        before they are sent to the {{ service.name }} server.
-        """
-        return request, metadata
-
-    {% if not method.void %}
-    {% if not method.server_streaming %}
-    def post_{{ method.name|snake_case }}(self, response: {{method.output.ident}}) -> {{method.output.ident}}:
-    {% else %}
-    def post_{{ method.name|snake_case }}(self, response: rest_streaming.ResponseIterator) -> rest_streaming.ResponseIterator:
-    {% endif %}
-        """Post-rpc interceptor for {{ method.name|snake_case }}
-
-        Override in a subclass to manipulate the response
-        after it is returned by the {{ service.name }} server but before
-        it is returned to user code.
-        """
-        return response
-    {% endif %}
-    {% endfor %}
-
-    {% for name, signature in api.mixin_api_signatures.items() %}
-    def pre_{{ name|snake_case }}(
-        self, request: {{signature.request_type}}, metadata: Sequence[Tuple[str, str]]
-    ) -> Tuple[{{signature.request_type}}, Sequence[Tuple[str, str]]]:
-        """Pre-rpc interceptor for {{ name|snake_case }}
-
-        Override in a subclass to manipulate the request or metadata
-        before they are sent to the {{ service.name }} server.
-        """
-        return request, metadata
-
-    def post_{{ name|snake_case }}(
-        self, response: {{signature.response_type}}
-    ) -> {{signature.response_type}}:
-        """Post-rpc interceptor for {{ name|snake_case }}
-
-        Override in a subclass to manipulate the response
-        after it is returned by the {{ service.name }} server but before
-        it is returned to user code.
-        """
-        return response
-    {% endfor %}
+{{ shared_macros.create_interceptor_class(api, service, method, is_async=False) }}
 
 
 @dataclasses.dataclass