feat: add grpc transcoding + tests

google/api_core/path_template.py

@@ -64,7 +64,7 @@ def _expand_variable_match(positional_vars, named_vars, match):
     """Expand a matched variable with its value.
 
     Args:
-        positional_vars (list): A list of positonal variables. This list will
+        positional_vars (list): A list of positional variables. This list will
             be modified.
         named_vars (dict): A dictionary of named variables.
         match (re.Match): A regular expression match.
@@ -193,3 +193,61 @@ def validate(tmpl, path):
     """
     pattern = _generate_pattern_for_template(tmpl) + "$"
     return True if re.match(pattern, path) is not None else False
+
+
+def transcode(http_options, **request_kwargs):
+    """Transcodes a grpc request pattern into a proper HTTP request following the rules outlined here,
+       https://github.com/googleapis/googleapis/blob/master/google/api/http.proto#L44-L312
+
+        Args:
+            http_options (list(dict)): A list of dicts which consist of these keys,
+                'method'    (str): The http method
+                'uri'       (str): The path template
+                'body'      (str): The body field name (optional)
+                (This is a simplified representation of the proto option `google.api.http`)
+            request_kwargs (dict) : A dict representing the request object
+
+        Returns:
+            dict: The transcoded request with these keys,
+                'method'        (str)   : The http method
+                'uri'           (str)   : The expanded uri
+                'body'          (dict)  : A dict representing the body (optional)
+                'query_params'  (dict)  : A dict mapping query parameter variables and values
+
+        Raises:
+            ValueError: If the request does not match the given template.
+    """
+    answer = {}
+    for http_option in http_options:
+
+        # Assign path
+        uri_template = http_option["uri"]
+        path_fields = [
+            match.group("name") for match in _VARIABLE_RE.finditer(uri_template)
+        ]
+        path_args = {field: request_kwargs.get(field, None) for field in path_fields}
+        leftovers = {k: v for k, v in request_kwargs.items() if k not in path_args}
+        answer["uri"] = expand(uri_template, **path_args)
+
+        if not validate(uri_template, answer["uri"]) or not all(path_args.values()):
+            continue
+
+        # Assign body and query params
+        body = http_option.get("body")
+
+        if body:
+            if body == "*":
+                answer["body"] = leftovers
+                answer["query_params"] = {}
+            else:
+                try:
+                    answer["body"] = leftovers.pop(body)
+                except KeyError:
+                    continue
+                answer["query_params"] = leftovers
+        else:
+            answer["query_params"] = leftovers
+        answer["method"] = http_option["method"]
+        return answer
+
+    raise ValueError("Request obj does not match any template")

tests/unit/test_path_template.py

@@ -113,3 +113,161 @@ def test__replace_variable_with_pattern():
     match.group.return_value = None
     with pytest.raises(ValueError, match="Unknown"):
         path_template._replace_variable_with_pattern(match)
+
+
+@pytest.mark.parametrize(
+    "http_options, request_kwargs, expected_result",
+    [
+        [
+            [["get", "/v1/no/template", ""]],
+            {"foo": "bar"},
+            ["get", "/v1/no/template", {}, {"foo": "bar"}],
+        ],
+        # Single templates
+        [
+            [["get", "/v1/{field}", ""]],
+            {"field": "parent"},
+            ["get", "/v1/parent", {}, {}],
+        ],
+        [
+            [["get", "/v1/{field.sub}", ""]],
+            {"field.sub": "parent", "foo": "bar"},
+            ["get", "/v1/parent", {}, {"foo": "bar"}],
+        ],
+        # Single segment wildcard
+        [
+            [["get", "/v1/{field=*}", ""]],
+            {"field": "parent"},
+            ["get", "/v1/parent", {}, {}],
+        ],
+        [
+            [["get", "/v1/{field=a/*/b/*}", ""]],
+            {"field": "a/parent/b/child", "foo": "bar"},
+            ["get", "/v1/a/parent/b/child", {}, {"foo": "bar"}],
+        ],
+        # Double segment wildcard
+        [
+            [["get", "/v1/{field=**}", ""]],
+            {"field": "parent/p1"},
+            ["get", "/v1/parent/p1", {}, {}],
+        ],
+        [
+            [["get", "/v1/{field=a/**/b/**}", ""]],
+            {"field": "a/parent/p1/b/child/c1", "foo": "bar"},
+            ["get", "/v1/a/parent/p1/b/child/c1", {}, {"foo": "bar"}],
+        ],
+        # Combined single and double segment wildcard
+        [
+            [["get", "/v1/{field=a/*/b/**}", ""]],
+            {"field": "a/parent/b/child/c1"},
+            ["get", "/v1/a/parent/b/child/c1", {}, {}],
+        ],
+        [
+            [["get", "/v1/{field=a/**/b/*}/v2/{name}", ""]],
+            {"field": "a/parent/p1/b/child", "name": "first", "foo": "bar"},
+            ["get", "/v1/a/parent/p1/b/child/v2/first", {}, {"foo": "bar"}],
+        ],
+        # Single field body
+        [
+            [["post", "/v1/no/template", "data"]],
+            {"data": {"id": 1, "info": "some info"}, "foo": "bar"},
+            ["post", "/v1/no/template", {"id": 1, "info": "some info"}, {"foo": "bar"}],
+        ],
+        [
+            [["post", "/v1/{field=a/*}/b/{name=**}", "data"]],
+            {
+                "field": "a/parent",
+                "name": "first/last",
+                "data": {"id": 1, "info": "some info"},
+                "foo": "bar",
+            },
+            [
+                "post",
+                "/v1/a/parent/b/first/last",
+                {"id": 1, "info": "some info"},
+                {"foo": "bar"},
+            ],
+        ],
+        # Wildcard body
+        [
+            [["post", "/v1/{field=a/*}/b/{name=**}", "*"]],
+            {
+                "field": "a/parent",
+                "name": "first/last",
+                "data": {"id": 1, "info": "some info"},
+                "foo": "bar",
+            },
+            [
+                "post",
+                "/v1/a/parent/b/first/last",
+                {"data": {"id": 1, "info": "some info"}, "foo": "bar"},
+                {},
+            ],
+        ],
+        # Additional bindings
+        [
+            [
+                ["post", "/v1/{field=a/*}/b/{name=**}", "extra_data"],
+                ["post", "/v1/{field=a/*}/b/{name=**}", "*"],
+            ],
+            {
+                "field": "a/parent",
+                "name": "first/last",
+                "data": {"id": 1, "info": "some info"},
+                "foo": "bar",
+            },
+            [
+                "post",
+                "/v1/a/parent/b/first/last",
+                {"data": {"id": 1, "info": "some info"}, "foo": "bar"},
+                {},
+            ],
+        ],
+        [
+            [
+                ["get", "/v1/{field=a/*}/b/{name=**}", ""],
+                ["get", "/v1/{field=a/*}/b/first/last", ""],
+            ],
+            {"field": "a/parent", "foo": "bar"},
+            ["get", "/v1/a/parent/b/first/last", {}, {"foo": "bar"}],
+        ],
+    ],
+)
+def test_transcode(http_options, request_kwargs, expected_result):
+    http_options, expected_result = helper_test_transcode(http_options, expected_result)
+    result = path_template.transcode(http_options, **request_kwargs)
+    assert result == expected_result
+
+
+@pytest.mark.parametrize(
+    "http_options, request_kwargs",
+    [
+        [[["get", "/v1/{name}", ""]], {"foo": "bar"}],
+        [[["get", "/v1/{name}", ""]], {"name": "first/last"}],
+        [[["get", "/v1/{name=mr/*/*}", ""]], {"name": "first/last"}],
+        [[["post", "/v1/{name}", "data"]], {"name": "first/last"}],
+    ],
+)
+def test_transcode_fails(http_options, request_kwargs):
+    http_options, _ = helper_test_transcode(http_options, range(4))
+    with pytest.raises(ValueError):
+        path_template.transcode(http_options, **request_kwargs)
+
+
+def helper_test_transcode(http_options_list, expected_result_list):
+    http_options = []
+    for opt_list in http_options_list:
+        http_option = {"method": opt_list[0], "uri": opt_list[1]}
+        if opt_list[2]:
+            http_option["body"] = opt_list[2]
+        http_options.append(http_option)
+
+    expected_result = {
+        "method": expected_result_list[0],
+        "uri": expected_result_list[1],
+        "query_params": expected_result_list[3],
+    }
+    if expected_result_list[2]:
+        expected_result["body"] = expected_result_list[2]
+
+    return (http_options, expected_result)