feat: Add REST Interceptors to support reading metadata (#2299)

Author: parthea

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

@@ -383,13 +383,36 @@ class {{ async_method_name_prefix }}{{ service.name }}RestInterceptor:
     {% endif %}
         """Post-rpc interceptor for {{ method.name|snake_case }}
 
-        Override in a subclass to manipulate the response
+        DEPRECATED. Please use the `post_{{ method.name|snake_case }}_with_metadata`
+        interceptor instead.
+
+        Override in a subclass to read or manipulate the response
         after it is returned by the {{ service.name }} server but before
-        it is returned to user code.
+        it is returned to user code. This `post_{{ method.name|snake_case }}` interceptor runs
+        before the `post_{{ method.name|snake_case }}_with_metadata` interceptor.
         """
         return response
 
+    {% if not method.server_streaming %}
+    {{ async_prefix }}def post_{{ method.name|snake_case }}_with_metadata(self, response: {{method.output.ident}}, {{ client_method_metadata_argument() }}) -> Tuple[{{method.output.ident}}, {{ client_method_metadata_type() }}]:
+    {% else %}
+    {{ async_prefix }}def post_{{ method.name|snake_case }}_with_metadata(self, response: rest_streaming{{ async_suffix }}.{{ async_method_name_prefix }}ResponseIterator, {{ client_method_metadata_argument() }}) -> Tuple[rest_streaming{{ async_suffix }}.{{ async_method_name_prefix }}ResponseIterator, {{ client_method_metadata_type() }}]:
     {% endif %}
+        """Post-rpc interceptor for {{ method.name|snake_case }}
+
+        Override in a subclass to read or manipulate the response or metadata after it
+        is returned by the {{ service.name }} server but before it is returned to user code.
+        
+        We recommend only using this `post_{{ method.name|snake_case }}_with_metadata`
+        interceptor in new development instead of the `post_{{ method.name|snake_case }}` interceptor.
+        When both interceptors are used, this `post_{{ method.name|snake_case }}_with_metadata` interceptor runs after the
+        `post_{{ method.name|snake_case }}` interceptor. The (possibly modified) response returned by 
+        `post_{{ method.name|snake_case }}` will be passed to
+        `post_{{ method.name|snake_case }}_with_metadata`.
+        """
+        return response, metadata
+
+    {% endif %}{# not method.void #}
     {% endfor %}
 
     {% for name, signature in api.mixin_api_signatures.items() %}

packages/gapic-generator/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

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest_asyncio.py.j2

@@ -217,6 +217,8 @@ class Async{{service.name}}RestTransport(_Base{{ service.name }}RestTransport):
             json_format.Parse(content, pb_resp, ignore_unknown_fields=True)
             {% endif %}{# if method.server_streaming #}
             resp = await self._interceptor.post_{{ method.name|snake_case }}(resp)
+            response_metadata = [(k, str(v)) for k, v in response.headers.items()]
+            resp, _ = await 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

packages/gapic-generator/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 #}