diff --git a/python-lib/dku_utils.py b/python-lib/dku_utils.py index a24cbde..692300a 100644 --- a/python-lib/dku_utils.py +++ b/python-lib/dku_utils.py @@ -306,3 +306,16 @@ def get_user_secrets(configuration): user_secrets[secret.get("key")] = secret.get("value") return user_secrets return {} + + +def join_url(base_url, segment): + # urllib.parse.urljoin can't be used here: + # urllib.parse.urljoin("https://services.odata.org/V3/Northwind/Northwind.svc", "Order_Details?$skiptoken=10436,75") + # -> https://services.odata.org/V3/Northwind/Order_Details?$skiptoken=10436,75 -> 404 + base_url = base_url or "" + base_url = base_url.strip("/") + segments = [base_url] + if segment: + segment = segment.lstrip("/") + segments.append(segment) + return "/".join(segments) diff --git a/python-lib/pagination.py b/python-lib/pagination.py index 84467fe..252c6b6 100644 --- a/python-lib/pagination.py +++ b/python-lib/pagination.py @@ -1,5 +1,5 @@ from safe_logger import SafeLogger -from dku_utils import get_value_from_path, extract_key_using_json_path +from dku_utils import get_value_from_path, extract_key_using_json_path, join_url logger = SafeLogger("api-connect plugin Pagination") @@ -110,7 +110,7 @@ def update_next_page_link(self, data, response_links=None): if self.next_page_key: next_page_path = extract_key_using_json_path(data, self.next_page_key) if self.next_page_url_base and next_page_path: - self.next_page_url = "/".join([self.next_page_url_base, next_page_path]) + self.next_page_url = join_url(self.next_page_url_base, next_page_path) else: self.next_page_url = next_page_path logger.info("update_next_page_link:next_page_url_base={}, next_page_path={}, next_page_url={}".format( diff --git a/tests/python/unit/test_common.py b/tests/python/unit/test_common.py index 0630771..ddc6e6f 100644 --- a/tests/python/unit/test_common.py +++ b/tests/python/unit/test_common.py @@ -1,4 +1,4 @@ -from dku_utils import template_dict +from dku_utils import template_dict, join_url import pytest @@ -32,3 +32,17 @@ def setup_class(self): def test_template_dict(self): endpoint = template_dict(self.template, **self.kwargs) assert endpoint == self.endpoint_ok + + def test_join_url(self): + assert "https://bla.com/bla/bli" == join_url("https://bla.com", "bla/bli") + assert "https://bla.com/bla/bli" == join_url("https://bla.com/", "bla/bli") + assert "https://bla.com/bla/bli" == join_url("https://bla.com/", "/bla/bli") + assert "https://bla.com/bla/bli" == join_url("https://bla.com/", "bla/bli") + assert "https://bla.com/bla/bli/" == join_url("https://bla.com", "bla/bli/") + assert "https://bla.com/bla/bli/" == join_url("https://bla.com/", "bla/bli/") + assert "https://bla.com/bla/bli/" == join_url("https://bla.com/", "/bla/bli/") + assert "https://bla.com/bla/bli/" == join_url("https://bla.com/", "bla/bli/") + assert "https://bla.com" == join_url("https://bla.com/", "") + assert "https://bla.com" == join_url("https://bla.com/", None) + assert "https://bla.com" == join_url("https://bla.com", "") + assert "https://bla.com" == join_url("https://bla.com", None)