Skip to content
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

feat: Add helper function to format query_params for rest transport. #275

Merged
merged 7 commits into from
Sep 20, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 94 additions & 0 deletions google/api_core/rest_helpers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
# Copyright 2021 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Helpers for rest transports."""

import functools
import operator


def flatten_query_params(obj):
"""Flatten a nested dict into a list of (name,value) tuples.

The result is suitable for setting query params on an http request.

.. code-block:: python

>>> obj = {'a':
... {'b':
... {'c': ['x', 'y', 'z']} },
... 'd': 'uvw', }
>>> flatten_query_params(obj)
[('a.b.c', 'x'), ('a.b.c', 'y'), ('a.b.c', 'z'), ('d', 'uvw')]

Note that, as described in
https://github.com/googleapis/googleapis/blob/48d9fb8c8e287c472af500221c6450ecd45d7d39/google/api/http.proto#L117,
repeated fields (i.e. list-valued fields) may only contain primitive types (not lists or dicts).
This is enforced in this function.

Args:
obj: a nested dictionary (from json), or None

Returns: a list of tuples, with each tuple having a (possibly) multi-part name
and a scalar value.

Raises:
TypeError if obj is not a dict or None
ValueError if obj contains a list of non-primitive values.
"""

if obj is not None and not isinstance(obj, dict):
raise TypeError("flatten_query_params must be called with dict object")

return _flatten(obj, key_path=[])


def _flatten(obj, key_path):
if obj is None:
return []
if isinstance(obj, dict):
return _flatten_dict(obj, key_path=key_path)
if isinstance(obj, list):
return _flatten_list(obj, key_path=key_path)
return _flatten_value(obj, key_path=key_path)


def _is_primitive_value(obj):
if obj is None:
return False

if isinstance(obj, (list, dict)):
raise ValueError("query params may not contain repeated dicts or lists")

return True


def _flatten_value(obj, key_path):
return [(".".join(key_path), obj)]


def _flatten_dict(obj, key_path):
items = (_flatten(value, key_path=key_path + [key]) for key, value in obj.items())
return functools.reduce(operator.concat, items, [])


def _flatten_list(elems, key_path):
# Only lists of scalar values are supported.
# The name (key_path) is repeated for each value.
items = (
_flatten_value(elem, key_path=key_path)
for elem in elems
if _is_primitive_value(elem)
)
return functools.reduce(operator.concat, items, [])
77 changes: 77 additions & 0 deletions tests/unit/test_rest_helpers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
# Copyright 2021 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import pytest

from google.api_core import rest_helpers


def test_flatten_simple_value():
with pytest.raises(TypeError):
rest_helpers.flatten_query_params("abc")


def test_flatten_list():
with pytest.raises(TypeError):
rest_helpers.flatten_query_params(["abc", "def"])


def test_flatten_none():
assert rest_helpers.flatten_query_params(None) == []


def test_flatten_empty_dict():
assert rest_helpers.flatten_query_params({}) == []


def test_flatten_simple_dict():
assert rest_helpers.flatten_query_params({"a": "abc", "b": "def"}) == [
("a", "abc"),
("b", "def"),
]


def test_flatten_repeated_field():
assert rest_helpers.flatten_query_params({"a": ["x", "y", "z", None]}) == [
("a", "x"),
("a", "y"),
("a", "z"),
]


def test_flatten_nested_dict():
obj = {"a": {"b": {"c": ["x", "y", "z"]}}, "d": {"e": "uvw"}}
expected_result = [("a.b.c", "x"), ("a.b.c", "y"), ("a.b.c", "z"), ("d.e", "uvw")]

assert rest_helpers.flatten_query_params(obj) == expected_result


def test_flatten_repeated_dict():
obj = {
"a": {"b": {"c": [{"v": 1}, {"v": 2}]}},
"d": "uvw",
}

with pytest.raises(ValueError):
rest_helpers.flatten_query_params(obj)


def test_flatten_repeated_list():
obj = {
"a": {"b": {"c": [["e", "f"], ["g", "h"]]}},
"d": "uvw",
}

with pytest.raises(ValueError):
rest_helpers.flatten_query_params(obj)