diff --git a/felt_python/api.py b/felt_python/api.py index 68c91d1..73fbf2d 100644 --- a/felt_python/api.py +++ b/felt_python/api.py @@ -4,6 +4,7 @@ import json as json_ import os import typing +import urllib.parse import urllib.request from importlib.metadata import version, PackageNotFoundError @@ -20,6 +21,36 @@ BASE_URL = os.getenv("FELT_BASE_URL", "https://felt.com/api/v2/") +def build_url(template: str, **path_params) -> str: + """Fill a URL template, percent-encoding each value as one path segment. + + Ids reach these functions from user code, config files and other API + responses, so they cannot be assumed URL-safe. Interpolating them directly + means an id containing a space raises http.client.InvalidURL before the + request is sent, and one containing "?", "#" or "/" silently changes the + path or query the server sees. Encoding each value with safe="" keeps a + bad id a plain 404. + + >>> build_url(BASE_URL + "maps/{map_id}", map_id="a b/c") + 'https://felt.com/api/v2/maps/a%20b%2Fc' + """ + return template.format( + **{ + key: urllib.parse.quote(str(value), safe="") + for key, value in path_params.items() + } + ) + + +def build_query(url: str, **params) -> str: + """Append a percent-encoded query string, skipping None values.""" + present = {key: value for key, value in params.items() if value is not None} + if not present: + return url + separator = "&" if "?" in url else "?" + return f"{url}{separator}{urllib.parse.urlencode(present)}" + + def make_request( url: str, method: typing.Literal["GET", "POST", "PATCH", "DELETE"], diff --git a/felt_python/comments.py b/felt_python/comments.py index 9152b2d..94f8aa5 100644 --- a/felt_python/comments.py +++ b/felt_python/comments.py @@ -4,7 +4,7 @@ from urllib.parse import urljoin -from .api import make_request, BASE_URL +from .api import BASE_URL, build_query, build_url, make_request COMMENT = urljoin(BASE_URL, "maps/{map_id}/comments/{comment_id}") @@ -23,7 +23,7 @@ def export_comments(map_id: str, format: str = "json", api_token: str | None = N Returns: The exported comments in the specified format """ - url = f"{COMMENT_EXPORT.format(map_id=map_id)}?format={format}" + url = build_query(build_url(COMMENT_EXPORT, map_id=map_id), format=format) response = make_request( url=url, method="GET", @@ -44,7 +44,7 @@ def resolve_comment(map_id: str, comment_id: str, api_token: str | None = None): Confirmation of the resolved comment """ response = make_request( - url=COMMENT_RESOLVE.format(map_id=map_id, comment_id=comment_id), + url=build_url(COMMENT_RESOLVE, map_id=map_id, comment_id=comment_id), method="POST", api_token=api_token, ) @@ -60,7 +60,7 @@ def delete_comment(map_id: str, comment_id: str, api_token: str | None = None): api_token: Optional API token """ make_request( - url=COMMENT.format(map_id=map_id, comment_id=comment_id), + url=build_url(COMMENT, map_id=map_id, comment_id=comment_id), method="DELETE", api_token=api_token, ) diff --git a/felt_python/elements.py b/felt_python/elements.py index fcd86dd..cacfb58 100644 --- a/felt_python/elements.py +++ b/felt_python/elements.py @@ -4,7 +4,7 @@ from urllib.parse import urljoin -from .api import make_request, BASE_URL +from .api import BASE_URL, build_url, make_request from .util import deprecated @@ -25,7 +25,7 @@ def list_elements(map_id: str, api_token: str | None = None): GeoJSON FeatureCollection of all elements """ response = make_request( - url=ELEMENTS.format(map_id=map_id), + url=build_url(ELEMENTS, map_id=map_id), method="GET", api_token=api_token, ) @@ -43,7 +43,7 @@ def list_element_groups(map_id: str, api_token: str | None = None): List of element groups """ response = make_request( - url=ELEMENT_GROUPS.format(map_id=map_id), + url=build_url(ELEMENT_GROUPS, map_id=map_id), method="GET", api_token=api_token, ) @@ -62,7 +62,7 @@ def get_element_group(map_id: str, element_group_id: str, api_token: str | None GeoJSON FeatureCollection of all elements in the group """ response = make_request( - url=ELEMENT_GROUP.format(map_id=map_id, element_group_id=element_group_id), + url=build_url(ELEMENT_GROUP, map_id=map_id, element_group_id=element_group_id), method="GET", api_token=api_token, ) @@ -106,7 +106,7 @@ def upsert_elements( "geojson_feature_collection must be a valid GeoJSON" ) response = make_request( - url=ELEMENTS.format(map_id=map_id), + url=build_url(ELEMENTS, map_id=map_id), method="POST", json=geojson_feature_collection, api_token=api_token, @@ -123,7 +123,7 @@ def delete_element(map_id: str, element_id: str, api_token: str | None = None): api_token: Optional API token """ make_request( - url=ELEMENT.format(map_id=map_id, element_id=element_id), + url=build_url(ELEMENT, map_id=map_id, element_id=element_id), method="DELETE", api_token=api_token, ) @@ -159,7 +159,7 @@ def upsert_element_groups( The created or updated element groups """ response = make_request( - url=ELEMENT_GROUPS.format(map_id=map_id), + url=build_url(ELEMENT_GROUPS, map_id=map_id), method="POST", json=element_groups, api_token=api_token, diff --git a/felt_python/layer_groups.py b/felt_python/layer_groups.py index 4dfdbfc..851f3fc 100644 --- a/felt_python/layer_groups.py +++ b/felt_python/layer_groups.py @@ -4,7 +4,7 @@ from urllib.parse import urljoin -from .api import make_request, BASE_URL +from .api import BASE_URL, build_url, make_request GROUPS = urljoin(BASE_URL, "maps/{map_id}/layer_groups") @@ -25,7 +25,7 @@ def list_layer_groups(map_id: str, api_token: str | None = None): List of layer groups """ response = make_request( - url=GROUPS.format(map_id=map_id), + url=build_url(GROUPS, map_id=map_id), method="GET", api_token=api_token, ) @@ -48,7 +48,7 @@ def get_layer_group( Layer group details """ response = make_request( - url=GROUP.format(map_id=map_id, layer_group_id=layer_group_id), + url=build_url(GROUP, map_id=map_id, layer_group_id=layer_group_id), method="GET", api_token=api_token, ) @@ -73,7 +73,7 @@ def update_layer_groups( The updated layer groups """ response = make_request( - url=GROUPS.format(map_id=map_id), + url=build_url(GROUPS, map_id=map_id), method="POST", json=layer_group_params_list, api_token=api_token, @@ -94,7 +94,7 @@ def delete_layer_group( api_token: Optional API token """ make_request( - url=GROUP.format(map_id=map_id, layer_group_id=layer_group_id), + url=build_url(GROUP, map_id=map_id, layer_group_id=layer_group_id), method="DELETE", api_token=api_token, ) @@ -136,7 +136,7 @@ def update_layer_group( json_payload["visibility_interaction"] = visibility_interaction response = make_request( - url=GROUP.format(map_id=map_id, layer_group_id=layer_group_id), + url=build_url(GROUP, map_id=map_id, layer_group_id=layer_group_id), method="POST", json=json_payload, api_token=api_token, @@ -166,7 +166,7 @@ def publish_layer_group( json_payload["name"] = name response = make_request( - url=GROUPS_PUBLISH.format(map_id=map_id, layer_group_id=layer_group_id), + url=build_url(GROUPS_PUBLISH, map_id=map_id, layer_group_id=layer_group_id), method="POST", json=json_payload, api_token=api_token, diff --git a/felt_python/layers.py b/felt_python/layers.py index 35f6d02..ff969cc 100644 --- a/felt_python/layers.py +++ b/felt_python/layers.py @@ -10,7 +10,7 @@ from urllib.parse import urljoin -from .api import make_request, BASE_URL +from .api import BASE_URL, build_url, make_request from .util import deprecated @@ -31,7 +31,7 @@ def list_layers(map_id: str, api_token: str | None = None): """List layers on a map""" response = make_request( - url=LAYERS.format(map_id=map_id), + url=build_url(LAYERS, map_id=map_id), method="GET", api_token=api_token, ) @@ -79,7 +79,7 @@ def upload_file( json_payload["zoom"] = zoom response = make_request( - url=LAYER_UPLOAD.format(map_id=map_id), + url=build_url(LAYER_UPLOAD, map_id=map_id), method="POST", api_token=api_token, json=json_payload, @@ -147,7 +147,7 @@ def refresh_file_layer( The refresh response including presigned upload details """ response = make_request( - url=LAYER_REFRESH.format(map_id=map_id, layer_id=layer_id), + url=build_url(LAYER_REFRESH, map_id=map_id, layer_id=layer_id), method="POST", api_token=api_token, ) @@ -186,7 +186,7 @@ def upload_url( json_payload["hints"] = hints response = make_request( - url=LAYER_UPLOAD.format(map_id=map_id), + url=build_url(LAYER_UPLOAD, map_id=map_id), method="POST", api_token=api_token, json=json_payload, @@ -197,7 +197,8 @@ def upload_url( def refresh_url_layer(map_id: str, layer_id: str, api_token: str | None = None): """Refresh a layer originated from a URL upload""" response = make_request( - url=LAYER_REFRESH.format( + url=build_url( + LAYER_REFRESH, map_id=map_id, layer_id=layer_id, ), @@ -219,7 +220,8 @@ def get_layer( ): """Get details of a layer""" response = make_request( - url=LAYER.format( + url=build_url( + LAYER, map_id=map_id, layer_id=layer_id, ), @@ -237,7 +239,8 @@ def update_layer_style( ): """Update a layer's style""" response = make_request( - url=LAYER_UPDATE_STYLE.format( + url=build_url( + LAYER_UPDATE_STYLE, map_id=map_id, layer_id=layer_id, ), @@ -258,7 +261,7 @@ def get_export_link( Vector layers will be downloaded in GPKG format. Raster layers will be GeoTIFFs. """ response = make_request( - url=LAYER_EXPORT_LINK.format(map_id=map_id, layer_id=layer_id), + url=build_url(LAYER_EXPORT_LINK, map_id=map_id, layer_id=layer_id), method="GET", api_token=api_token, ) @@ -306,7 +309,7 @@ def update_layers( The updated layers """ response = make_request( - url=LAYERS.format(map_id=map_id), + url=build_url(LAYERS, map_id=map_id), method="POST", json=layer_params_list, api_token=api_token, @@ -321,7 +324,7 @@ def delete_layer( ): """Delete a layer from a map""" make_request( - url=LAYER.format(map_id=map_id, layer_id=layer_id), + url=build_url(LAYER, map_id=map_id, layer_id=layer_id), method="DELETE", api_token=api_token, ) @@ -349,7 +352,7 @@ def publish_layer( json_payload["name"] = name response = make_request( - url=LAYER_PUBLISH.format(map_id=map_id, layer_id=layer_id), + url=build_url(LAYER_PUBLISH, map_id=map_id, layer_id=layer_id), method="POST", json=json_payload, api_token=api_token, @@ -389,7 +392,7 @@ def create_custom_export( json_payload["filters"] = filters response = make_request( - url=LAYER_CUSTOM_EXPORT.format(map_id=map_id, layer_id=layer_id), + url=build_url(LAYER_CUSTOM_EXPORT, map_id=map_id, layer_id=layer_id), method="POST", json=json_payload, api_token=api_token, @@ -415,7 +418,8 @@ def get_custom_export_status( Export status including download URL when complete """ response = make_request( - url=LAYER_CUSTOM_EXPORT_STATUS.format( + url=build_url( + LAYER_CUSTOM_EXPORT_STATUS, map_id=map_id, layer_id=layer_id, export_id=export_id, diff --git a/felt_python/library.py b/felt_python/library.py index 26aa0b9..839f666 100644 --- a/felt_python/library.py +++ b/felt_python/library.py @@ -4,7 +4,7 @@ from urllib.parse import urljoin -from .api import make_request, BASE_URL +from .api import BASE_URL, build_query, make_request LIBRARY = urljoin(BASE_URL, "library") @@ -24,7 +24,7 @@ def list_library_layers(source: str = "workspace", api_token: str | None = None) Returns: The layer library containing layers and layer groups """ - url = f"{LIBRARY}?source={source}" + url = build_query(LIBRARY, source=source) response = make_request( url=url, method="GET", diff --git a/felt_python/maps.py b/felt_python/maps.py index 904a10f..5af3a55 100644 --- a/felt_python/maps.py +++ b/felt_python/maps.py @@ -4,7 +4,7 @@ from urllib.parse import urljoin -from .api import make_request, BASE_URL +from .api import BASE_URL, build_query, build_url, make_request from .util import deprecated @@ -89,7 +89,7 @@ def create_map( def delete_map(map_id: str, api_token: str | None = None): """Delete a map""" make_request( - url=MAP.format(map_id=map_id), + url=build_url(MAP, map_id=map_id), method="DELETE", api_token=api_token, ) @@ -98,7 +98,7 @@ def delete_map(map_id: str, api_token: str | None = None): def get_map(map_id: str, api_token: str | None = None): """Get details of a map""" response = make_request( - url=MAP.format(map_id=map_id), + url=build_url(MAP, map_id=map_id), method="GET", api_token=api_token, ) @@ -162,7 +162,7 @@ def update_map( json_args["viewer_permissions"] = viewer_permissions response = make_request( - url=MAP_UPDATE.format(map_id=map_id), + url=build_url(MAP_UPDATE, map_id=map_id), method="POST", json=json_args, api_token=api_token, @@ -199,7 +199,7 @@ def move_map( json_args["folder_id"] = folder_id response = make_request( - url=MAP_MOVE.format(map_id=map_id), + url=build_url(MAP_MOVE, map_id=map_id), method="POST", json=json_args, api_token=api_token, @@ -222,9 +222,7 @@ def create_embed_token( Returns: The created embed token with expiration time """ - url = MAP_EMBED_TOKEN.format(map_id=map_id) - if user_email: - url = f"{url}?user_email={user_email}" + url = build_query(build_url(MAP_EMBED_TOKEN, map_id=map_id), user_email=user_email) response = make_request( url=url, @@ -252,7 +250,7 @@ def add_source_layer( Acceptance status and links to the created resources """ response = make_request( - url=MAP_ADD_SOURCE_LAYER.format(map_id=map_id), + url=build_url(MAP_ADD_SOURCE_LAYER, map_id=map_id), method="POST", json=source_layer_params, api_token=api_token, @@ -294,7 +292,7 @@ def duplicate_map( json_args["destination"] = {"folder_id": folder_id} response = make_request( - url=MAP_DUPLICATE.format(map_id=map_id), + url=build_url(MAP_DUPLICATE, map_id=map_id), method="POST", json=json_args, api_token=api_token, diff --git a/felt_python/projects.py b/felt_python/projects.py index 6285197..34f5e04 100644 --- a/felt_python/projects.py +++ b/felt_python/projects.py @@ -4,7 +4,7 @@ from urllib.parse import urljoin -from .api import make_request, BASE_URL +from .api import BASE_URL, build_query, build_url, make_request PROJECTS = urljoin(BASE_URL, "projects/") @@ -14,9 +14,7 @@ def list_projects(workspace_id: str | None = None, api_token: str | None = None): """List all projects accessible to the authenticated user""" - url = PROJECTS - if workspace_id: - url = f"{url}?workspace_id={workspace_id}" + url = build_query(PROJECTS, workspace_id=workspace_id) response = make_request( url=url, method="GET", @@ -49,7 +47,7 @@ def create_project(name: str, visibility: str, api_token: str | None = None): def get_project(project_id: str, api_token: str | None = None): """Get details of a project""" response = make_request( - url=PROJECT.format(project_id=project_id), + url=build_url(PROJECT, project_id=project_id), method="GET", api_token=api_token, ) @@ -80,7 +78,7 @@ def update_project( json_args["visibility"] = visibility response = make_request( - url=PROJECT_UPDATE.format(project_id=project_id), + url=build_url(PROJECT_UPDATE, project_id=project_id), method="POST", json=json_args, api_token=api_token, @@ -94,7 +92,7 @@ def delete_project(project_id: str, api_token: str | None = None): Note: This will delete all Folders and Maps inside the project! """ make_request( - url=PROJECT.format(project_id=project_id), + url=build_url(PROJECT, project_id=project_id), method="DELETE", api_token=api_token, ) diff --git a/felt_python/sources.py b/felt_python/sources.py index 1317b7e..90433f5 100644 --- a/felt_python/sources.py +++ b/felt_python/sources.py @@ -4,7 +4,7 @@ from urllib.parse import urljoin -from .api import make_request, BASE_URL +from .api import BASE_URL, build_query, build_url, make_request SOURCES = urljoin(BASE_URL, "sources") @@ -15,9 +15,7 @@ def list_sources(workspace_id: str | None = None, api_token: str | None = None): """List all sources accessible to the authenticated user""" - url = SOURCES - if workspace_id: - url = f"{url}?workspace_id={workspace_id}" + url = build_query(SOURCES, workspace_id=workspace_id) response = make_request( url=url, method="GET", @@ -59,7 +57,7 @@ def create_source( def get_source(source_id: str, api_token: str | None = None): """Get details of a source""" response = make_request( - url=SOURCE.format(source_id=source_id), + url=build_url(SOURCE, source_id=source_id), method="GET", api_token=api_token, ) @@ -94,7 +92,7 @@ def update_source( json_payload["permissions"] = permissions response = make_request( - url=SOURCE_UPDATE.format(source_id=source_id), + url=build_url(SOURCE_UPDATE, source_id=source_id), method="POST", json=json_payload, api_token=api_token, @@ -105,7 +103,7 @@ def update_source( def delete_source(source_id: str, api_token: str | None = None): """Delete a source""" make_request( - url=SOURCE.format(source_id=source_id), + url=build_url(SOURCE, source_id=source_id), method="DELETE", api_token=api_token, ) @@ -118,7 +116,7 @@ def sync_source(source_id: str, api_token: str | None = None): The source reference with synchronization status """ response = make_request( - url=SOURCE_SYNC.format(source_id=source_id), + url=build_url(SOURCE_SYNC, source_id=source_id), method="POST", api_token=api_token, ) diff --git a/tests/tests.py b/tests/tests.py index d258d91..facc78b 100644 --- a/tests/tests.py +++ b/tests/tests.py @@ -17,6 +17,7 @@ from projects_test import FeltProjectsTest from sources_test import FeltSourcesTest from delete_test import FeltDeleteTest +from url_building_test import BuildQueryTest, BuildUrlTest if __name__ == "__main__": @@ -32,6 +33,8 @@ # Add all test classes test_cases = [ + BuildUrlTest, + BuildQueryTest, FeltAPITest, FeltElementsTest, FeltLayersTest, diff --git a/tests/url_building_test.py b/tests/url_building_test.py new file mode 100644 index 0000000..7e02813 --- /dev/null +++ b/tests/url_building_test.py @@ -0,0 +1,99 @@ +""" +Unit tests for URL construction. + +Unlike the rest of the suite these need no API token and make no requests. +""" + +import os +import sys +import unittest + +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from felt_python.api import build_query, build_url +from felt_python.comments import COMMENT +from felt_python.layers import LAYER +from felt_python.maps import MAP + + +class BuildUrlTest(unittest.TestCase): + def test_ordinary_ids_are_untouched(self): + """Felt's short slugs and UUIDs contain nothing that needs encoding.""" + for map_id in ( + "PF0ve5FaSWujSB5402D8wD", + "deadbeef-0000-0000-0000-000000000000", + ): + with self.subTest(map_id=map_id): + self.assertTrue( + build_url(MAP, map_id=map_id).endswith(f"/maps/{map_id}") + ) + + def test_space_is_encoded_instead_of_raising(self): + """Previously http.client raised InvalidURL before sending anything.""" + self.assertTrue( + build_url(MAP, map_id="not a real id").endswith("/maps/not%20a%20real%20id") + ) + + def test_query_and_fragment_cannot_escape_the_path(self): + """An id with "?" or "#" used to silently truncate the path.""" + self.assertTrue(build_url(MAP, map_id="abc?x=1").endswith("/maps/abc%3Fx%3D1")) + self.assertTrue(build_url(MAP, map_id="abc#frag").endswith("/maps/abc%23frag")) + + def test_slash_cannot_add_path_segments(self): + """safe="" is the point: a "/" in an id must not become a separator.""" + url = build_url(MAP, map_id="../../sources") + self.assertTrue(url.endswith("/maps/..%2F..%2Fsources"), url) + + def test_each_segment_is_encoded_independently(self): + url = build_url(LAYER, map_id="a b", layer_id="c/d") + self.assertTrue(url.endswith("/maps/a%20b/layers/c%2Fd"), url) + + def test_unicode_ids(self): + self.assertTrue(build_url(MAP, map_id="mapa-ñ").endswith("/maps/mapa-%C3%B1")) + + def test_non_string_values_are_coerced(self): + self.assertTrue(build_url(MAP, map_id=123).endswith("/maps/123")) + + def test_multi_segment_template(self): + url = build_url(COMMENT, map_id="m1", comment_id="c1") + self.assertTrue(url.endswith("/maps/m1/comments/c1"), url) + + +class BuildQueryTest(unittest.TestCase): + def test_none_values_are_omitted(self): + self.assertEqual( + build_query("https://x/api", workspace_id=None), "https://x/api" + ) + + def test_single_param(self): + self.assertEqual( + build_query("https://x/api", workspace_id="w1"), + "https://x/api?workspace_id=w1", + ) + + def test_values_are_encoded(self): + """A "+" in an email would otherwise decode server-side as a space.""" + self.assertEqual( + build_query("https://x/api", user_email="a+b@example.com"), + "https://x/api?user_email=a%2Bb%40example.com", + ) + + def test_ampersand_cannot_inject_a_parameter(self): + self.assertEqual( + build_query("https://x/api", source="felt&admin=true"), + "https://x/api?source=felt%26admin%3Dtrue", + ) + + def test_appends_to_an_existing_query_string(self): + self.assertEqual( + build_query("https://x/api?a=1", b="2"), "https://x/api?a=1&b=2" + ) + + def test_mixed_present_and_absent(self): + self.assertEqual( + build_query("https://x/api", a="1", b=None), "https://x/api?a=1" + ) + + +if __name__ == "__main__": + unittest.main(verbosity=2)