Fetch subsequent paginated data on list_public_trades

Signed-off-by: camille-bouvy-frequenz <[email protected]>

Author: camille-bouvy-frequenz

integration_tests/test_api.py

@@ -347,14 +347,26 @@ async def test_list_gridpool_trades(set_up: dict[str, Any]) -> None:
 
 
 @pytest.mark.asyncio
-async def test_list_public_trades(set_up: dict[str, Any]) -> None:
+async def test_list_public_trades_pages(set_up: dict[str, Any]) -> None:
     """Test listing public trades."""
-    public_trades = await set_up["client"].list_public_trades(
-        delivery_period=set_up["delivery_period"],
-        max_nr_trades=10,
-    )
-    assert len(public_trades) >= 0
+    public_trades = []
+    page_limit = 3  
+    page_count = 0
 
+    delivery_period = DeliveryPeriod(
+        start=datetime.fromisoformat("2024-06-10T10:00:00+00:00"),
+        duration=timedelta(minutes=15)
+    )  
+    async for page in set_up["client"].list_public_trades_pages(
+        delivery_period=delivery_period,
+        page_size=5,
+    ):
+        public_trades.extend(page)
+        page_count += 1
+        if page_count >= page_limit:
+            break
+
+    assert len(public_trades) == 15, "Public trades page count mismatch"
 
 @pytest.mark.asyncio
 async def test_stream_gridpool_orders(set_up: dict[str, Any]) -> None:

src/frequenz/client/electricity_trading/_client.py

@@ -8,9 +8,10 @@
 import logging
 from datetime import datetime, timezone
 from decimal import Decimal, InvalidOperation
-from typing import Awaitable, cast
+from typing import AsyncIterator, Awaitable, cast
 
 import grpc
+from frequenz.api.common.v1.pagination.pagination_params_pb2 import PaginationParams
 
 # pylint: disable=no-member
 from frequenz.api.electricity_trading.v1 import (
@@ -857,29 +858,27 @@ async def list_gridpool_trades(
             _logger.exception("Error occurred while listing gridpool trades: %s", e)
             raise
 
-    async def list_public_trades(
+    async def list_public_trades_pages(
         # pylint: disable=too-many-arguments, too-many-positional-arguments
         self,
         states: list[TradeState] | None = None,
         delivery_period: DeliveryPeriod | None = None,
         buy_delivery_area: DeliveryArea | None = None,
         sell_delivery_area: DeliveryArea | None = None,
-        max_nr_trades: int | None = None,
-        page_token: str | None = None,
-    ) -> list[PublicTrade]:
+        page_size: int | None = None,
+    ) -> AsyncIterator[list[PublicTrade]]:
         """
-        List all executed public orders with optional filters.
+        List all executed public orders with optional filters and pagination.
 
         Args:
             states: List of order states to filter by.
             delivery_period: The delivery period to filter by.
             buy_delivery_area: The buy delivery area to filter by.
             sell_delivery_area: The sell delivery area to filter by.
-            max_nr_trades: The maximum number of trades to return.
-            page_token: The page token to use for pagination.
+            page_size: The number of public trades to return per page.
 
-        Returns:
-            The list of public trades.
+        Yields:
+            The list of public trades for each page.
 
         Raises:
             grpc.RpcError: If an error occurs while listing public trades.
@@ -891,27 +890,36 @@ async def list_public_trades(
             sell_delivery_area=sell_delivery_area,
         )
 
-        pagination_params = Params(
-            page_size=max_nr_trades,
-            page_token=page_token,
+        request = electricity_trading_pb2.ListPublicTradesRequest(
+            filter=public_trade_filter.to_pb(),
+            pagination_params=(
+                Params(page_size=page_size).to_proto() if page_size else None
+            ),
         )
 
-        try:
-            response = await cast(
-                Awaitable[electricity_trading_pb2.ListPublicTradesResponse],
-                self.stub.ListPublicTrades(
-                    electricity_trading_pb2.ListPublicTradesRequest(
-                        filter=public_trade_filter.to_pb(),
-                        pagination_params=pagination_params.to_proto(),
-                    ),
-                    metadata=self._metadata,
-                ),
-            )
+        while True:
+            try:
+                response = await cast(
+                    Awaitable[electricity_trading_pb2.ListPublicTradesResponse],
+                    self.stub.ListPublicTrades(request, metadata=self._metadata),
+                )
 
-            return [
-                PublicTrade.from_pb(public_trade)
-                for public_trade in response.public_trades
-            ]
-        except grpc.RpcError as e:
-            _logger.exception("Error occurred while listing public trades: %s", e)
-            raise
+                public_trades = [
+                    PublicTrade.from_pb(public_trade)
+                    for public_trade in response.public_trades
+                ]
+
+                yield public_trades
+
+                if len(response.pagination_info.next_page_token):
+                    request.pagination_params.CopyFrom(
+                        PaginationParams(
+                            page_token=response.pagination_info.next_page_token
+                        )
+                    )
+                else:
+                    break
+
+            except grpc.RpcError as e:
+                _logger.exception("Error occurred while listing public trades: %s", e)
+                raise