fix(api-core): move http encoding into transcode method

packages/google-api-core/google/api_core/path_template.py

@@ -123,7 +123,7 @@ def _expand_variable_match(positional_vars, named_vars, match):
         try:
             val = str(named_vars[name])
             _extract_and_validate_wildcards(val, template, name)
-            return urllib.parse.quote(val, safe="/")
+            return val
         except KeyError:
             raise ValueError(
                 "Named variable '{}' not specified and needed by template "
@@ -133,7 +133,7 @@ def _expand_variable_match(positional_vars, named_vars, match):
         try:
             val = str(positional_vars.pop(0))
             _extract_and_validate_wildcards(val, positional, "positional variable")
-            return urllib.parse.quote(val, safe="/")
+            return val
         except IndexError:
             raise ValueError(
                 "Positional variable not specified and needed by template "
@@ -215,12 +215,17 @@ def _generate_pattern_for_template(tmpl):
     return _VARIABLE_RE.sub(_replace_variable_with_pattern, tmpl)
 
 
-def get_field(request, field):
+def get_field(request, field, encode=False):
     """Get the value of a field from a given dictionary.
 
     Args:
         request (dict | Message): A dictionary or a Message object.
         field (str): The key to the request in dot notation.
+        encode (bool): Whether to percent-encode the field value. If enabled,
+            will encode all characters except `[-_.~/0-9a-zA-Z]` for URI path
+            variable parts per
+            https://github.com/googleapis/googleapis/blob/master/google/api/http.proto#L44-L312.
+            Defaults to False.
 
     Returns:
         The value of the field.
@@ -235,6 +240,8 @@ def get_field(request, field):
             value = value.get(part)
     if isinstance(value, dict):
         return
+    if encode and value is not None:
+        return urllib.parse.quote(str(value), safe="/")
     return value
 
 
@@ -327,7 +334,10 @@ def transcode(http_options, message=None, **request_kwargs):
         ]
         bindings.append((uri_template, fields))
 
-        path_args = {field: get_field(transcoded_value, field) for field, _ in fields}
+        path_args = {
+            field: get_field(transcoded_value, field, encode=True)
+            for field, _ in fields
+        }
         request["uri"] = expand(uri_template, **path_args)
 
         if not validate(uri_template, request["uri"]) or not all(path_args.values()):

packages/google-api-core/tests/unit/test_path_template.py

@@ -18,7 +18,6 @@
 
 import pytest
 from google.api import auth_pb2
-
 from google.api_core import path_template
 
 
@@ -64,28 +63,27 @@
             {"name": "parent/child/object"},
             "/v1/a/parent/child/object",
         ],
-        # Encoding / Metacharacters in positional and named params
-        ["/v1/*", ["..?$httpMethod=DELETE#"], {}, "/v1/..%3F%24httpMethod%3DDELETE%23"],
-        ["/v1/**", ["path/sub/with/?and#"], {}, "/v1/path/sub/with/%3Fand%23"],
+        # Legitimate resource name patterns preserved in expand
         [
-            "/v1/{name}",
+            "projects/{project}/databases/{database}",
             [],
-            {"name": "..?$httpMethod=DELETE#"},
-            "/v1/..%3F%24httpMethod%3DDELETE%23",
+            {"project": "my-project", "database": "(default)"},
+            "projects/my-project/databases/(default)",
         ],
         [
-            "/v1/{name=**}",
+            "/v1/{name}",
             [],
-            {"name": "path/sub/with/?and#"},
-            "/v1/path/sub/with/%3Fand%23",
+            {"name": "my-instance:cluster-1"},
+            "/v1/my-instance:cluster-1",
         ],
         [
-            "/v3/{session=projects/*/locations/*/agents/*/sessions/*}:detectIntent",
+            "projects/{project}/serviceAccounts/{account}",
             [],
             {
-                "session": "projects/cx/locations/global/agents/a1/sessions/..?$httpMethod=DELETE#"
+                "project": "my-project",
+                "account": "service@my-project.iam.gserviceaccount.com",
             },
-            "/v3/projects/cx/locations/global/agents/a1/sessions/..%3F%24httpMethod%3DDELETE%23:detectIntent",
+            "projects/my-project/serviceAccounts/service@my-project.iam.gserviceaccount.com",
         ],
     ],
 )
@@ -133,6 +131,18 @@ def test_get_field(request_obj, field, expected_result):
     assert result == expected_result
 
 
+def test_get_field_encode():
+    assert (
+        path_template.get_field({"name": "a$b?c=d#e f"}, "name", encode=True)
+        == "a%24b%3Fc%3Dd%23e%20f"
+    )
+    assert (
+        path_template.get_field({"name": "abc-._~"}, "name", encode=True) == "abc-._~"
+    )
+    assert path_template.get_field({"name": None}, "name", encode=True) is None
+    assert path_template.get_field({}, "name", encode=True) is None
+
+
 @pytest.mark.parametrize(
     "request_obj, field, expected_result",
     [
@@ -417,6 +427,43 @@ def test_transcode_subfields(http_options, message, request_kwargs, expected_res
                 auth_pb2.AuthenticationRule(oauth=auth_pb2.OAuthRequirements()),
             ],
         ],
+        # Special characters, colons, and parentheses in URI variable transcoding
+        [
+            [["get", "/v1/projects/{p}/databases/{d}", ""]],
+            None,
+            {"p": "proj", "d": "(default)", "foo": "bar"},
+            ["get", "/v1/projects/proj/databases/%28default%29", {}, {"foo": "bar"}],
+        ],
+        [
+            [["get", "/v1/{name}", ""]],
+            None,
+            {"name": "inst:1", "foo": "bar"},
+            ["get", "/v1/inst%3A1", {}, {"foo": "bar"}],
+        ],
+        [
+            [["get", "/v1/{name}", ""]],
+            None,
+            {"name": "a$b?c=d#e f", "foo": "bar"},
+            ["get", "/v1/a%24b%3Fc%3Dd%23e%20f", {}, {"foo": "bar"}],
+        ],
+        [
+            [["get", "/v1/{name}", ""]],
+            None,
+            {"name": "..?$httpMethod=DELETE#", "foo": "bar"},
+            ["get", "/v1/..%3F%24httpMethod%3DDELETE%23", {}, {"foo": "bar"}],
+        ],
+        [
+            [["get", "/v1/{name=**}", ""]],
+            None,
+            {"name": "sub/?and#", "foo": "bar"},
+            ["get", "/v1/sub/%3Fand%23", {}, {"foo": "bar"}],
+        ],
+        [
+            [["post", "/v1/{name=a/*}:verb", ""]],
+            None,
+            {"name": "a/..?$httpMethod=DELETE#", "foo": "bar"},
+            ["post", "/v1/a/..%3F%24httpMethod%3DDELETE%23:verb", {}, {"foo": "bar"}],
+        ],
     ],
 )
 def test_transcode_with_wildcard(
@@ -792,23 +839,6 @@ def test_path_traversal_dots_validation_double_star_invalid(name_val):
         )
 
 
-@pytest.mark.parametrize(
-    "tmpl, kwargs, expected_result",
-    [
-        ["/v1/{name}", {"name": "abc-._~"}, "/v1/abc-._~"],
-        ["/v1/{name=**}", {"name": "abc-._~/"}, "/v1/abc-._~/"],
-        ["/v1/{name}", {"name": "a/b"}, "/v1/a/b"],
-        ["/v1/{name}", {"name": "a$b?c=d#e f"}, "/v1/a%24b%3Fc%3Dd%23e%20f"],
-    ],
-)
-def test_percent_encoding_unreserved_characters(tmpl, kwargs, expected_result):
-    result = path_template.expand(tmpl, **kwargs)
-    assert result == expected_result
-    # For single-segment with '/', validate should fail because '/' is preserved
-    if "/" in kwargs.get("name", "") and tmpl == "/v1/{name}":
-        assert not path_template.validate(tmpl, result)
-
-
 @pytest.mark.parametrize(
     "tmpl, args, kwargs, expected_err_match",
     [