Skip to content

feat: Allow representing enums with their unqualified symbolic names in headers #465

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 12 commits into from
Nov 10, 2022
Merged
20 changes: 17 additions & 3 deletions google/api_core/gapic_v1/routing_header.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,38 +20,52 @@
Generally, these headers are specified as gRPC metadata.
"""

from enum import Enum
from urllib.parse import urlencode

ROUTING_METADATA_KEY = "x-goog-request-params"


def to_routing_header(params):
def to_routing_header(params, qualified_enums=True):
"""Returns a routing header string for the given request parameters.

Args:
params (Mapping[str, Any]): A dictionary containing the request
parameters used for routing.
qualified_enums (bool): Whether to represent enum values
as their type-qualified symbol names instead of as their
unqualified symbol names.

Returns:
str: The routing header string.

"""
if not qualified_enums:
if isinstance(params, dict):
tuples = params.items()
else:
tuples = params
params = [(x[0], x[1].name) if isinstance(x[1], Enum) else x for x in tuples]
return urlencode(
params,
# Per Google API policy (go/api-url-encoding), / is not encoded.
safe="/",
)


def to_grpc_metadata(params):
def to_grpc_metadata(params, qualified_enums=True):
"""Returns the gRPC metadata containing the routing headers for the given
request parameters.

Args:
params (Mapping[str, Any]): A dictionary containing the request
parameters used for routing.
qualified_enums (bool): Whether to represent enum values
as their type-qualified symbol names instead of as their
unqualified symbol names.

Returns:
Tuple(str, str): The gRPC metadata containing the routing header key
and value.
"""
return (ROUTING_METADATA_KEY, to_routing_header(params))
return (ROUTING_METADATA_KEY, to_routing_header(params, qualified_enums))
1 change: 0 additions & 1 deletion noxfile.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,6 @@ def default(session, install_grpc=True):
CURRENT_DIRECTORY / "testing" / f"constraints-{session.python}.txt"
)

# Install all test dependencies, then install this package in-place.
session.install(
"dataclasses",
"mock",
Expand Down
2 changes: 0 additions & 2 deletions owlbot.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,8 +48,6 @@
""",
)

s.replace(".github/workflows/lint.yml", "python-version: \"3.10\"", "python-version: \"3.7\"")

python.configure_previous_major_version_branches()

s.shell.run(["nox", "-s", "blacken"], hide_output=False)
31 changes: 31 additions & 0 deletions tests/unit/gapic/test_routing_header.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.

from enum import Enum

import pytest

try:
Expand All @@ -35,6 +37,35 @@ def test_to_routing_header_with_slashes():
assert value == "name=me/ep&book.read=1%262"


def test_enum_fully_qualified():
class Message:
class Color(Enum):
RED = 1
GREEN = 2
BLUE = 3

params = [("color", Message.Color.RED)]
value = routing_header.to_routing_header(params)
assert value == "color=Color.RED"
value = routing_header.to_routing_header(params, qualified_enums=True)
assert value == "color=Color.RED"


def test_enum_nonqualified():
class Message:
class Color(Enum):
RED = 1
GREEN = 2
BLUE = 3

params = [("color", Message.Color.RED), ("num", 5)]
value = routing_header.to_routing_header(params, qualified_enums=False)
assert value == "color=RED&num=5"
params = {"color": Message.Color.RED, "num": 5}
value = routing_header.to_routing_header(params, qualified_enums=False)
assert value == "color=RED&num=5"


def test_to_grpc_metadata():
params = [("name", "meep"), ("book.read", "1")]
metadata = routing_header.to_grpc_metadata(params)
Expand Down