feat: Support alternative http bindings in the gapic schema.

gapic/schema/wrappers.py

@@ -36,6 +36,7 @@
 from google.api import annotations_pb2      # type: ignore
 from google.api import client_pb2
 from google.api import field_behavior_pb2
+from google.api import http_pb2
 from google.api import resource_pb2
 from google.api_core import exceptions      # type: ignore
 from google.protobuf import descriptor_pb2  # type: ignore
@@ -821,13 +822,64 @@ def field_headers(self) -> Sequence[str]:
 
         return next((tuple(pattern.findall(verb)) for verb in potential_verbs if verb), ())
 
+    def http_rule_to_tuple(self, http_rule: http_pb2.HttpRule) -> Tuple[str, str, str]:
+        """Represent salient info in an http rule as a tuple.
+
+        Args:
+          http_rule: the http option message to examine.
+
+        Returns:
+          A tuple of (method, uri pattern, body or None),
+            or None if no method is specified.
+        """
+
+        http_dict: Mapping[str, str]
+
+        method = http_rule.WhichOneof('pattern')
+        if method is None or method == 'custom':
+            return ('', '', '')
+
+        uri = getattr(http_rule, method)
+        if not uri:
+            return ('', '', '')
+        body = http_rule.body if http_rule.body else None
+        return (method, uri, body)
+
+    @property
+    def http_options(self) -> List[Dict[str, str]]:
+        """Return a list of the http options for this method.
+
+        e.g. [{'method': 'post'
+              'uri': '/some/path'
+              'body': '*'},]
+
+        """
+        http = self.options.Extensions[annotations_pb2.http]
+        # shallow copy is fine here (elements are not modified)
+        http_options = list(http.additional_bindings)
+        # Main pattern comes first
+        http_options.insert(0, http)
+        answers: List[Dict[str, str]] = []
+
+        for http_rule in http_options:
+            method, uri, body = self.http_rule_to_tuple(http_rule)
+            if not method:
+                continue
+            answer: Dict[str, str] = {}
+            answer['method'] = method
+            answer['uri'] = uri
+            if body:
+                answer['body'] = body
+            answers.append(answer)
+        return answers
+
     @property
     def http_opt(self) -> Optional[Dict[str, str]]:
-        """Return the http option for this method.
+        """Return the (main) http option for this method.
 
-        e.g. {'verb': 'post'
-              'url': '/some/path'
-              'body': '*'}
+          e.g. {'verb': 'post'
+                'url': '/some/path'
+                'body': '*'}
 
         """
         http: List[Tuple[descriptor_pb2.FieldDescriptorProto, str]]

tests/unit/schema/wrappers/test_method.py

@@ -328,6 +328,85 @@ def test_method_path_params_no_http_rule():
     assert method.path_params == []
 
 
+def test_method_http_options():
+    verbs = [
+        'get',
+        'put',
+        'post',
+        'delete',
+        'patch'
+    ]
+    for v in verbs:
+        http_rule = http_pb2.HttpRule(**{v: '/v1/{parent=projects/*}/topics'})
+        method = make_method('DoSomething', http_rule=http_rule)
+        assert method.http_options == [{
+            'method': v,
+            'uri': '/v1/{parent=projects/*}/topics'
+        }]
+
+
+def test_method_http_options_empty_http_rule():
+    http_rule = http_pb2.HttpRule()
+    method = make_method('DoSomething', http_rule=http_rule)
+    assert method.http_options == []
+
+    http_rule = http_pb2.HttpRule(get='')
+    method = make_method('DoSomething', http_rule=http_rule)
+    assert method.http_options == []
+
+
+def test_method_http_options_no_http_rule():
+    method = make_method('DoSomething')
+    assert method.path_params == []
+
+
+def test_method_http_options_body():
+    http_rule = http_pb2.HttpRule(
+        post='/v1/{parent=projects/*}/topics',
+        body='*'
+    )
+    method = make_method('DoSomething', http_rule=http_rule)
+    assert method.http_options == [{
+        'method': 'post',
+        'uri': '/v1/{parent=projects/*}/topics',
+        'body': '*'
+    }]
+
+
+def test_method_http_options_additional_bindings():
+    http_rule = http_pb2.HttpRule(
+        post='/v1/{parent=projects/*}/topics',
+        body='*',
+        additional_bindings=[
+            http_pb2.HttpRule(
+                post='/v1/{parent=projects/*/regions/*}/topics',
+                body='*',
+            ),
+            http_pb2.HttpRule(
+                post='/v1/projects/p1/topics',
+                body='body_field',
+            ),
+        ]
+    )
+    method = make_method('DoSomething', http_rule=http_rule)
+    assert len(method.http_options) == 3
+    assert {
+        'method': 'post',
+        'uri': '/v1/{parent=projects/*}/topics',
+        'body': '*'
+    } in method.http_options
+    assert {
+        'method': 'post',
+        'uri': '/v1/{parent=projects/*/regions/*}/topics',
+        'body': '*'
+    } in method.http_options
+    assert {
+        'method': 'post',
+        'uri': '/v1/projects/p1/topics',
+        'body': 'body_field'
+    } in method.http_options
+
+
 def test_method_query_params():
     # tests only the basic case of grpc transcoding
     http_rule = http_pb2.HttpRule(