diff --git a/ibm_cloud_sdk_core/__init__.py b/ibm_cloud_sdk_core/__init__.py index 04c6413..e5da1dd 100644 --- a/ibm_cloud_sdk_core/__init__.py +++ b/ibm_cloud_sdk_core/__init__.py @@ -29,6 +29,7 @@ string_to_date: De-serializes a string to a date. convert_model: Convert a model object into an equivalent dict. convert_list: Convert a list of strings into comma-separated string. + get_query_param: Return a query parameter value from a URL read_external_sources: Get config object from external sources. get_authenticator_from_environment: Get authenticator from external sources. """ @@ -42,4 +43,5 @@ from .utils import datetime_to_string, string_to_datetime, read_external_sources from .utils import date_to_string, string_to_date from .utils import convert_model, convert_list +from .utils import get_query_param from .get_authenticator import get_authenticator_from_environment diff --git a/ibm_cloud_sdk_core/utils.py b/ibm_cloud_sdk_core/utils.py index 87411e9..44185c4 100644 --- a/ibm_cloud_sdk_core/utils.py +++ b/ibm_cloud_sdk_core/utils.py @@ -19,6 +19,7 @@ from os import getenv, environ, getcwd from os.path import isfile, join, expanduser from typing import List, Union +from urllib.parse import urlparse, parse_qs import dateutil.parser as date_parser @@ -141,6 +142,29 @@ def string_to_date(string: str) -> datetime.date: """ return date_parser.parse(string).date() +def get_query_param(url_str: str, param: str) -> str: + """Return a query parameter value from url_str + + Args: + url_str: the URL from which to extract the query + parameter value + param: the name of the query parameter whose value + should be returned + + Returns: + the value of the `param` query parameter as a string + + Raises: + ValueError: if errors are encountered parsing `url_str` + """ + if not url_str: + return None + url = urlparse(url_str) + if not url.query: + return None + query = parse_qs(url.query, strict_parsing=True) + values = query.get(param) + return values[0] if values else None def convert_model(val: any) -> dict: """Convert a model object into an equivalent dict. diff --git a/test/test_utils.py b/test/test_utils.py index faeae42..184079c 100644 --- a/test/test_utils.py +++ b/test/test_utils.py @@ -17,11 +17,13 @@ import os import datetime - from typing import Optional +import pytest + from ibm_cloud_sdk_core import string_to_datetime, datetime_to_string, get_authenticator_from_environment from ibm_cloud_sdk_core import string_to_date, date_to_string from ibm_cloud_sdk_core import convert_model, convert_list +from ibm_cloud_sdk_core import get_query_param from ibm_cloud_sdk_core import read_external_sources from ibm_cloud_sdk_core.authenticators import BasicAuthenticator, IAMAuthenticator @@ -122,6 +124,46 @@ def test_date_conversion(): assert res == '2017-03-06' assert date_to_string(None) is None +def test_get_query_param(): + # Relative URL + next_url = '/api/v1/offerings?start=foo&limit=10' + page_token = get_query_param(next_url, 'start') + assert page_token == 'foo' + # Absolute URL + next_url = 'https://acme.com/api/v1/offerings?start=bar&limit=10' + page_token = get_query_param(next_url, 'start') + assert page_token == 'bar' + # Missing param + next_url = 'https://acme.com/api/v1/offerings?start=bar&limit=10' + page_token = get_query_param(next_url, 'token') + assert page_token is None + # No URL + page_token = get_query_param(None, 'start') + assert page_token is None + # Empty URL + page_token = get_query_param('', 'start') + assert page_token is None + # No query string + next_url = '/api/v1/offerings' + page_token = get_query_param(next_url, 'start') + assert page_token is None + # Bad query string + next_url = '/api/v1/offerings?start%XXfoo' + with pytest.raises(ValueError): + page_token = get_query_param(next_url, 'start') + # Duplicate param + next_url = '/api/v1/offerings?start=foo&start=bar&limit=10' + page_token = get_query_param(next_url, 'start') + assert page_token == 'foo' + # Bad URL - since the behavior for this case varies based on the version of Python + # we allow _either_ a ValueError or that the illegal chars are just ignored + next_url = 'https://foo.bar\u2100/api/v1/offerings?start=foo' + try: + page_token = get_query_param(next_url, 'start') + assert page_token == 'foo' + except ValueError: + # This is okay. + pass def test_convert_model(): class MockModel: