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
76 changes: 76 additions & 0 deletions google/api_core/rest_helpers.py
@@ -0,0 +1,76 @@
# 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 itertools


def flatten_query_params(obj, key_path=[]):
"""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')]

Args:
obj: a nested dictionary (from json)
key_path: a list of name segments, representing levels above this obj.

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

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_value(obj):
if obj is None:
return False
return not (isinstance(obj, list) or isinstance(obj, dict))
kbandes marked this conversation as resolved.
Show resolved Hide resolved


def _flatten_value(obj, key_path=[]):
kbandes marked this conversation as resolved.
Show resolved Hide resolved
if not key_path:
# There must be a key.
return []
kbandes marked this conversation as resolved.
Show resolved Hide resolved
return [('.'.join(key_path), obj)]


def _flatten_dict(obj, key_path=[]):
kbandes marked this conversation as resolved.
Show resolved Hide resolved
return list(
itertools.chain(*(flatten_query_params(v, key_path=key_path + [k])
kbandes marked this conversation as resolved.
Show resolved Hide resolved
for k, v in obj.items())))
kbandes marked this conversation as resolved.
Show resolved Hide resolved


def _flatten_list(l, key_path=[]):
kbandes marked this conversation as resolved.
Show resolved Hide resolved
# Only lists of scalar values are supported.
# The name (key_path) is repeated for each value.
return list(
itertools.chain(*(_flatten_value(elem, key_path=key_path)
for elem in l
kbandes marked this conversation as resolved.
Show resolved Hide resolved
if _is_value(elem))))
kbandes marked this conversation as resolved.
Show resolved Hide resolved
61 changes: 61 additions & 0 deletions tests/unit/test_rest_helpers.py
@@ -0,0 +1,61 @@
# 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.

from google.api_core import rest_helpers


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']}) == [
('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_ignore_repeated_dict():
obj = {'a':
{'b':
{'c':
[{'v': 1}, {'v': 2}]
}
},
'd': 'uvw', }
# a.b.c is a repeated dict - ignored
expected_result = [('d', 'uvw')]

assert rest_helpers.flatten_query_params(obj) == expected_result