add predicate-based find and filter helpers to API endpoint response Result model

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])

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