feat: Support alternative http bindings in the gapic schema. (#993)

Support alternative http bindings in the gapic schema and adds support for parsing multiple bindings for one method.

Co-authored-by: Kenneth Bandes <kbandes@google.com>

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
@@ -706,6 +707,27 @@ class RetryInfo:
     retryable_exceptions: FrozenSet[exceptions.GoogleAPICallError]
 
 
+@dataclasses.dataclass(frozen=True)
+class HttpRule:
+    """Representation of the method's http bindings."""
+    method: str
+    uri: str
+    body: Optional[str]
+
+    @classmethod
+    def try_parse_http_rule(cls, http_rule) -> Optional['HttpRule']:
+        method = http_rule.WhichOneof("pattern")
+        if method is None or method == "custom":
+            return None
+
+        uri = getattr(http_rule, method)
+        if not uri:
+            return None
+
+        body = http_rule.body or None
+        return cls(method, uri, body)
+
+
 @dataclasses.dataclass(frozen=True)
 class Method:
     """Description of a method (defined with the ``rpc`` keyword)."""
@@ -821,13 +843,22 @@ def field_headers(self) -> Sequence[str]:
 
         return next((tuple(pattern.findall(verb)) for verb in potential_verbs if verb), ())
 
+    @property
+    def http_options(self) -> List[HttpRule]:
+        """Return a list of the http bindings for this method."""
+        http = self.options.Extensions[annotations_pb2.http]
+        http_options = [http] + list(http.additional_bindings)
+        opt_gen = (HttpRule.try_parse_http_rule(http_rule)
+                   for http_rule in http_options)
+        return [rule for rule in opt_gen if rule]
+
     @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

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import collections
+import dataclasses
 from typing import Sequence
 
 from google.api import field_behavior_pb2
@@ -328,6 +329,86 @@ 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 [dataclasses.asdict(http) for http in method.http_options] == [{
+            'method': v,
+            'uri': '/v1/{parent=projects/*}/topics',
+            'body': None
+        }]
+
+
+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 [dataclasses.asdict(http) for http in 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 [dataclasses.asdict(http) for http in method.http_options] == [
+        {
+            'method': 'post',
+            'uri': '/v1/{parent=projects/*}/topics',
+            'body': '*'
+            },
+        {
+            'method': 'post',
+            'uri': '/v1/{parent=projects/*/regions/*}/topics',
+            'body': '*'
+            },
+        {
+            'method': 'post',
+            'uri': '/v1/projects/p1/topics',
+            'body': 'body_field'
+            }]
+
+
 def test_method_query_params():
     # tests only the basic case of grpc transcoding
     http_rule = http_pb2.HttpRule(