Skip to content
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
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.

### Added

- Added `ItemSearch.url_with_parameters` to get a formatted search url [#304](https://github.com/stac-utils/pystac-client/issues/304)

### Changed

### Fixed
Expand Down
59 changes: 45 additions & 14 deletions pystac_client/item_search.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import json
import re
import urllib.parse
import warnings
from collections.abc import Iterable, Mapping
from copy import deepcopy
Expand All @@ -22,6 +23,7 @@
from dateutil.relativedelta import relativedelta
from dateutil.tz import tzutc
from pystac import Collection, Item, ItemCollection
from requests import Request

from pystac_client._utils import Modifiable, call_modifier
from pystac_client.conformance import ConformanceClasses
Expand Down Expand Up @@ -311,23 +313,52 @@ def get_parameters(self) -> Dict[str, Any]:
if self.method == "POST":
return self._parameters
elif self.method == "GET":
params = deepcopy(self._parameters)
if "bbox" in params:
params["bbox"] = ",".join(map(str, params["bbox"]))
if "ids" in params:
params["ids"] = ",".join(params["ids"])
if "collections" in params:
params["collections"] = ",".join(params["collections"])
if "intersects" in params:
params["intersects"] = json.dumps(params["intersects"])
if "sortby" in params:
params["sortby"] = self._sortby_dict_to_str(params["sortby"])
if "fields" in params:
params["fields"] = self._fields_dict_to_str(params["fields"])
return params
return self._clean_params_for_get_request()
else:
raise Exception(f"Unsupported method {self.method}")

def _clean_params_for_get_request(self) -> Dict[str, Any]:
params = deepcopy(self._parameters)
if "bbox" in params:
params["bbox"] = ",".join(map(str, params["bbox"]))
if "ids" in params:
params["ids"] = ",".join(params["ids"])
if "collections" in params:
params["collections"] = ",".join(params["collections"])
if "intersects" in params:
params["intersects"] = json.dumps(params["intersects"])
if "sortby" in params:
params["sortby"] = self._sortby_dict_to_str(params["sortby"])
if "fields" in params:
params["fields"] = self._fields_dict_to_str(params["fields"])
return params

def url_with_parameters(self) -> str:
"""Returns this item search url with parameters, appropriate for a GET request.

Examples:

>>> search = ItemSearch(
... url="https://planetarycomputer.microsoft.com/api/stac/v1/search",
... collections=["cop-dem-glo-30"],
... bbox=[88.214, 27.927, 88.302, 28.034],
... )
>>> assert (
... search.url_with_parameters()
... == "https://planetarycomputer.microsoft.com/api/stac/v1/search?"
... "limit=100&bbox=88.214,27.927,88.302,28.034&collections=cop-dem-glo-30"
... )

Returns:
str: The search url with parameters.
"""
params = self._clean_params_for_get_request()
request = Request("GET", self.url, params=params)
url = request.prepare().url
if url is None:
raise ValueError("Could not construct a full url")
return urllib.parse.unquote(url)

def _format_query(self, value: Optional[QueryLike]) -> Optional[Dict[str, Any]]:
if value is None:
return None
Expand Down
21 changes: 21 additions & 0 deletions tests/test_item_search.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,27 @@ def bboxer() -> Iterator[float]:
search = ItemSearch(url=SEARCH_URL, bbox=bboxer())
assert search.get_parameters()["bbox"] == (-104.5, 44.0, -104.0, 45.0)

def test_url_with_parameters(self) -> None:
# Single timestamp input
search = ItemSearch(
url=SEARCH_URL,
datetime="2020-02-01T00:00:00Z",
bbox=[-104.5, 44.0, -104.0, 45.0],
)
assert "bbox=-104.5,44.0,-104.0,45.0" in search.url_with_parameters()

# Motivating example: https://github.com/stac-utils/pystac-client/issues/299
search = ItemSearch(
url="https://planetarycomputer.microsoft.com/api/stac/v1/search",
collections=["cop-dem-glo-30"],
bbox=[88.214, 27.927, 88.302, 28.034],
)
assert (
search.url_with_parameters()
== "https://planetarycomputer.microsoft.com/api/stac/v1/search?"
"limit=100&bbox=88.214,27.927,88.302,28.034&collections=cop-dem-glo-30"
)

def test_single_string_datetime(self) -> None:
# Single timestamp input
search = ItemSearch(url=SEARCH_URL, datetime="2020-02-01T00:00:00Z")
Expand Down