diff --git a/src/gfwapiclient/http/models/response.py b/src/gfwapiclient/http/models/response.py index 5702834..d9d35ab 100644 --- a/src/gfwapiclient/http/models/response.py +++ b/src/gfwapiclient/http/models/response.py @@ -2,7 +2,9 @@ from typing import ( Any, + Callable, Generic, + Iterator, List, Optional, Set, @@ -82,7 +84,7 @@ def data( The API endpoint result data, either a single `ResultItem` or a list of `ResultItem`. """ _items: Union[List[_ResultItemT], _ResultItemT] = ( - [*self._data] if isinstance(self._data, list) else self._data + list(self._data) if isinstance(self._data, list) else self._data ) return _items @@ -115,14 +117,97 @@ def df( A `DataFrame` representing the API endpoint result. If the result items contain geospatial data, a `GeoDataFrame` may be returned. """ - items: List[_ResultItemT] = ( - [*self._data] if isinstance(self._data, list) else [self._data] - ) df = pd.DataFrame( - [item.model_dump(include=include, exclude=exclude) for item in items], + [ + item.model_dump(include=include, exclude=exclude) + for item in self._iter_data() + ], **kwargs, ) return df + def filter( + self, + *, + predicate: Optional[Callable[[_ResultItemT], bool]] = None, + ) -> "Result[_ResultItemT]": + """Filters API endpoint result data using a predicate function. + + This method returns a new `Result` instance containing only those + `ResultItem` objects for which `predicate(item)` evaluates to `True`. + + If `predicate` is `None`, a shallow copy of the result is returned, + containing all `ResultItem` objects. + + Args: + predicate (Optional[Callable[[_ResultItemT], bool]], default=None): + An optional callable that accepts a `ResultItem` instance and + returns `True` if it should be included. + + Returns: + Result[_ResultItemT]: + A new `Result` instance containing the filtered `ResultItem` objects. + """ + if predicate is None or not callable(predicate): + return self.__class__(data=list(self._iter_data())) + + filtered_items: List[_ResultItemT] = [ + item for item in self._iter_data() if predicate(item) + ] + return self.__class__(data=filtered_items) + + def find( + self, + *, + predicate: Optional[Callable[[_ResultItemT], bool]] = None, + ) -> Optional[_ResultItemT]: + """Finds the first API endpoint result item matching a predicate. + + This method returns the first `ResultItem` for which + `predicate(item)` evaluates to `True`. + + If `predicate` is `None`, or if no items match, `None` is returned. + + Args: + predicate (Optional[Callable[[_ResultItemT], bool]], default=None): + An optional callable that accepts a `ResultItem` instance and + returns `True` for the desired item. + + Returns: + Optional[_ResultItemT]: + The first matching `ResultItem`, or `None` if no match is found. + """ + if predicate is None or not callable(predicate): + return None + + for item in self._iter_data(): + if predicate(item): + return item + + return None + + def _iter_data(self) -> Iterator[_ResultItemT]: + """Iterate lazily over API endpoint result data without copying. + + This internal helper provides a unified iteration interface over + the underlying response data regardless of whether the result + contains: + + - a single `ResultItem` + - a list of `ResultItem` + + Yields: + _ResultItemT: + Individual `ResultItem` contained in API endpoint result data. + + Returns: + Iterator[_ResultItemT]: + An iterator over API endpoint result data. + """ + if isinstance(self._data, list): + yield from self._data + else: + yield self._data + _ResultT = TypeVar("_ResultT", bound=Result[Any]) diff --git a/tests/http/test_response_result_model.py b/tests/http/test_response_result_model.py index dcd69e3..e2c2ac3 100644 --- a/tests/http/test_response_result_model.py +++ b/tests/http/test_response_result_model.py @@ -204,3 +204,245 @@ def test_result_dataframe_conversion_exclude( assert len(output) == 2 assert "id" not in list(output.columns) assert "flags" not in list(output.columns) + + +def test_result_filter_returns_only_items_matching_predicate( + mock_result_item: Dict[str, Any], +) -> None: + """Tests that `Result` filter with a predicate returns a new `Result` containing only matched `ResultItem` objects.""" + input = {**mock_result_item} + data = [SampleResultItem(**input), SampleResultItem(**{**input, "confidence": 4})] + result = SampleListResult(data=data) + + def predicate(item: SampleResultItem) -> bool: + return item is not None and item.confidence is not None and item.confidence >= 4 + + filtered_result: Result[SampleResultItem] = result.filter(predicate=predicate) + filtered_data: List[SampleResultItem] = cast( + List[SampleResultItem], filtered_result.data() + ) + + assert filtered_result is not result + assert isinstance(filtered_result, SampleListResult) + assert len(filtered_data) == 1 + assert filtered_data[-1].confidence == 4 + + +def test_result_filter_returns_empty_result_when_no_items_match_predicate( + mock_result_item: Dict[str, Any], +) -> None: + """Tests that `Result` filter with a predicate that matches no `ResultItem` objects returns a new empty `Result`.""" + input = {**mock_result_item} + data = [SampleResultItem(**input), SampleResultItem(**{**input, "confidence": 4})] + result = SampleListResult(data=data) + + def predicate(item: SampleResultItem) -> bool: + return item is not None and item.confidence is not None and item.confidence >= 5 + + filtered_result: Result[SampleResultItem] = result.filter(predicate=predicate) + filtered_data: List[SampleResultItem] = cast( + List[SampleResultItem], filtered_result.data() + ) + + assert filtered_result is not result + assert isinstance(filtered_result, SampleListResult) + assert len(filtered_data) == 0 + assert filtered_data == [] + + +def test_result_filter_without_predicate_returns_all_items( + mock_result_item: Dict[str, Any], +) -> None: + """Tests that `Result` filter without a predicate returns new `Result` containing all `ResultItem` objects.""" + input = {**mock_result_item} + data = [SampleResultItem(**input), SampleResultItem(**{**input, "confidence": 4})] + result = SampleListResult(data=data) + + filtered_result: Result[SampleResultItem] = result.filter() + filtered_data: List[SampleResultItem] = cast( + List[SampleResultItem], filtered_result.data() + ) + + assert filtered_result is not result + assert isinstance(filtered_result, SampleListResult) + assert len(filtered_data) == 2 + + +def test_result_filter_works_with_single_result_item( + mock_result_item: Dict[str, Any], +) -> None: + """Tests that `Result` filter works correctly with a single `ResultItem`.""" + input = {**mock_result_item} + data = SampleResultItem(**input) + result = SampleSingleResult(data=data) + + def predicate(item: SampleResultItem) -> bool: + return item is not None and item.confidence is not None and item.confidence >= 3 + + filtered_result: Result[SampleResultItem] = result.filter(predicate=predicate) + filtered_data: List[SampleResultItem] = cast( + List[SampleResultItem], filtered_result.data() + ) + + assert filtered_result is not result + assert isinstance(filtered_result, SampleSingleResult) + assert len(filtered_data) == 1 + assert filtered_data[-1].confidence == 3 + + +def test_result_filter_works_with_empty_result_set() -> None: + """Tests that `Result` filter works correctly with empty `ResultItem` objects.""" + result = SampleListResult(data=[]) + + def predicate(item: SampleResultItem) -> bool: + return item is not None and item.confidence is not None and item.confidence >= 3 + + filtered_result: Result[SampleResultItem] = result.filter(predicate=predicate) + filtered_data: List[SampleResultItem] = cast( + List[SampleResultItem], filtered_result.data() + ) + + assert filtered_result is not result + assert isinstance(filtered_result, SampleListResult) + assert len(filtered_data) == 0 + assert filtered_data == [] + + +@pytest.mark.parametrize( + "invalid_predicate", + [ + "invalid", + 123, + object(), + True, + [], + {}, + ], +) +def test_result_filter_invalid_predicate_returns_result_copy( + mock_result_item: Dict[str, Any], + invalid_predicate: Any, +) -> None: + """Tests that `Result` filter with an invalid predicates returns shalslow copy of the `Result`.""" + input = {**mock_result_item} + data = [SampleResultItem(**input), SampleResultItem(**{**input, "confidence": 4})] + result = SampleListResult(data=data) + + filtered_result: Result[SampleResultItem] = result.filter( + predicate=invalid_predicate + ) + filtered_data: List[SampleResultItem] = cast( + List[SampleResultItem], filtered_result.data() + ) + + assert filtered_result is not result + assert isinstance(filtered_result, SampleListResult) + assert len(filtered_data) == 2 + + +def test_result_find_returns_first_matching_item( + mock_result_item: Dict[str, Any], +) -> None: + """Tests that `Result` find returns the first `ResultItem` matching the predicate.""" + input = {**mock_result_item} + data = [SampleResultItem(**input), SampleResultItem(**{**input, "confidence": 4})] + result = SampleListResult(data=data) + + def predicate(item: SampleResultItem) -> bool: + return item.confidence is not None and item.confidence >= 3 + + found: Optional[SampleResultItem] = result.find(predicate=predicate) + + assert found is not None + assert isinstance(found, SampleResultItem) + assert found.confidence == 3 + + +def test_result_find_returns_none_when_no_items_match_predicate( + mock_result_item: Dict[str, Any], +) -> None: + """Tests that `Result` find returns `None` when no `ResultItem` matches the predicate.""" + input = {**mock_result_item} + data = [SampleResultItem(**input), SampleResultItem(**{**input, "confidence": 4})] + result = SampleListResult(data=data) + + def predicate(item: SampleResultItem) -> bool: + return item.confidence is not None and item.confidence >= 5 + + found: Optional[SampleResultItem] = result.find(predicate=predicate) + + assert found is None + + +def test_result_find_returns_none_when_predicate_is_none( + mock_result_item: Dict[str, Any], +) -> None: + """Tests that `Result` find returns `None` when predicate is not provided.""" + input = {**mock_result_item} + data = [SampleResultItem(**input)] + result = SampleListResult(data=data) + + found: Optional[SampleResultItem] = result.find() + + assert found is None + + +def test_result_find_returns_none_for_empty_result_set() -> None: + """Tests that `Result` find returns `None` when the result contains no items.""" + result = SampleListResult(data=[]) + + def predicate(item: SampleResultItem) -> bool: + return True + + found: Optional[SampleResultItem] = result.find(predicate=predicate) + + assert found is None + + +def test_result_find_works_with_single_result_item( + mock_result_item: Dict[str, Any], +) -> None: + """Tests that `Result` find works correctly with a single `ResultItem`.""" + item = SampleResultItem(**mock_result_item) + result = SampleSingleResult(data=item) + + def predicate(item: SampleResultItem) -> bool: + return item.id == cast(str, mock_result_item["id"]) + + found: Optional[SampleResultItem] = result.find(predicate=predicate) + + assert found is not None + assert found is item + + +def test_result_find_returns_none_when_single_item_does_not_match( + mock_result_item: Dict[str, Any], +) -> None: + """Tests that `Result` find returns `None` when the single `ResultItem` does not match.""" + item = SampleResultItem(**mock_result_item) + result = SampleSingleResult(data=item) + + def predicate(_: SampleResultItem) -> bool: + return False + + found: Optional[SampleResultItem] = result.find(predicate=predicate) + + assert found is None + + +@pytest.mark.parametrize( + "invalid_predicate", + ["invalid", 123, object(), True, [], {}], +) +def test_result_find_invalid_predicate_returns_none( + mock_result_item: Dict[str, Any], + invalid_predicate: Any, +) -> None: + """Tests that `Result` find with an invalid predicates returns `None`.""" + input = {**mock_result_item} + data = [SampleResultItem(**input)] + result = SampleListResult(data=data) + + found: Optional[SampleResultItem] = result.find(predicate=invalid_predicate) + + assert found is None