feat: Add REST Interceptors to support reading metadata

gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/_shared_macros.j2

@@ -356,7 +356,21 @@ class {{ async_method_name_prefix }}{{ service.name }}RestInterceptor:
         it is returned to user code.
         """
         return response
+
+    {% if not method.server_streaming %}
+    {{ async_prefix }}def post_{{ method.name|snake_case }}_with_metadata(self, response: {{method.output.ident}}, metadata: Sequence[Tuple[str, str]]) -> Tuple[{{method.output.ident}}, Sequence[Tuple[str, str]]]:
+    {% else %}
+    {{ async_prefix }}def post_{{ method.name|snake_case }}_with_metadata(self, response: rest_streaming{{ async_suffix }}.{{ async_method_name_prefix }}ResponseIterator, metadata: Sequence[Tuple[str, str]]) -> Tuple[rest_streaming{{ async_suffix }}.{{ async_method_name_prefix }}ResponseIterator, Sequence[Tuple[str, str]]]:
     {% endif %}
+        """Post-rpc interceptor for {{ method.name|snake_case }}
+
+        Override in a subclass to either manipulate or read, either the response
+        or metadata after it is returned by the {{ service.name }} server but before
+        it is returned to user code.
+        """
+        return response, metadata
+
+    {% endif %}{# not method.void #}
     {% endif %}{# if (not is_async) or (is_async and not method.lro and not method.extended_lro and not method.paged_result_field) #}
     {% endfor %}
 

gapic/ads-templates/%namespace/%name/%version/%sub/services/%service/transports/rest.py.j2

@@ -256,6 +256,8 @@ class {{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             json_format.Parse(response.content, pb_resp, ignore_unknown_fields=True)
             {% endif %}{# method.lro #}
             resp = self._interceptor.post_{{ method.name|snake_case }}(resp)
+            response_metadata = [(k, str(v)) for k, v in response.headers.items()]
+            resp, _ = self._interceptor.post_{{ method.name|snake_case }}_with_metadata(resp, response_metadata)
             return resp
 
             {% endif %}{# method.void #}

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

@@ -389,7 +389,20 @@ class {{ async_method_name_prefix }}{{ service.name }}RestInterceptor:
         """
         return response
 
+    {% if not method.server_streaming %}
+    {{ async_prefix }}def post_{{ method.name|snake_case }}_with_metadata(self, response: {{method.output.ident}}, metadata: Sequence[Tuple[str, str]]) -> Tuple[{{method.output.ident}}, Sequence[Tuple[str, str]]]:
+    {% else %}
+    {{ async_prefix }}def post_{{ method.name|snake_case }}_with_metadata(self, response: rest_streaming{{ async_suffix }}.{{ async_method_name_prefix }}ResponseIterator, metadata: Sequence[Tuple[str, str]]) -> Tuple[rest_streaming{{ async_suffix }}.{{ async_method_name_prefix }}ResponseIterator, Sequence[Tuple[str, str]]]:
     {% endif %}
+        """Post-rpc interceptor for {{ method.name|snake_case }}
+
+        Override in a subclass to either manipulate or read, either the response
+        or metadata after it is returned by the {{ service.name }} server but before
+        it is returned to user code.
+        """
+        return response, metadata
+
+    {% endif %}{# not method.void #}
     {% endfor %}
 
     {% for name, signature in api.mixin_api_signatures.items() %}

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

@@ -256,6 +256,8 @@ class {{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             {% endif %}{# method.lro #}
             {#- TODO(https://github.com/googleapis/gapic-generator-python/issues/2274): Add debug log before intercepting a request #} 
             resp = self._interceptor.post_{{ method.name|snake_case }}(resp)
+            response_metadata = [(k, str(v)) for k, v in response.headers.items()]
+            resp, _ = self._interceptor.post_{{ method.name|snake_case }}_with_metadata(resp, response_metadata)
             {# TODO(https://github.com/googleapis/gapic-generator-python/issues/2279): Add logging support for rest streaming. #}
             {% if not method.server_streaming %}
             if CLIENT_LOGGING_SUPPORTED and _LOGGER.isEnabledFor(logging.DEBUG):  # pragma: NO COVER

gapic/templates/tests/unit/gapic/%name_%version/%sub/test_macros.j2

@@ -2218,11 +2218,13 @@ def test_initialize_client_w_{{transport_name}}():
         {% endif %}
         {% if not method.void %}
         mock.patch.object(transports.{{async_method_prefix}}{{ service.name }}RestInterceptor, "post_{{method.name|snake_case}}") as post, \
+        mock.patch.object(transports.{{async_method_prefix}}{{ service.name }}RestInterceptor, "post_{{method.name|snake_case}}_with_metadata") as post_with_metadata, \
         {% endif %}
         mock.patch.object(transports.{{async_method_prefix}}{{ service.name }}RestInterceptor, "pre_{{ method.name|snake_case }}") as pre:
         pre.assert_not_called()
         {% if not method.void %}
         post.assert_not_called()
+        post_with_metadata.assert_not_called()
         {% endif %}
         {% if method.input.ident.is_proto_plus_type %}
         pb_message = {{ method.input.ident }}.pb({{ method.input.ident }}())
@@ -2265,13 +2267,15 @@ def test_initialize_client_w_{{transport_name}}():
         pre.return_value = request, metadata
         {% if not method.void %}
         post.return_value = {{ method.output.ident }}()
+        post_with_metadata.return_value = {{ method.output.ident }}(), metadata
         {% endif %}
 
         {{await_prefix}}client.{{ method_name }}(request, metadata=[("key", "val"), ("cephalopod", "squid"),])
 
         pre.assert_called_once()
         {% if not method.void %}
         post.assert_called_once()
+        post_with_metadata.assert_called_once()
         {% endif %}
 {% endif %}{# end 'grpc' in transport #}
 {% endmacro%}{# inteceptor_class_test #}