fix: pagination limit

src/pyfwapi/apiconnection.py

@@ -2,6 +2,8 @@
 import random
 import typing as t
 import urllib.parse
+from datetime import UTC
+from urllib.parse import quote
 
 import aiolimiter
 import httpxyz as httpx
@@ -27,6 +29,9 @@
 BACKOFF_BASE_SECONDS = 1.0
 BACKOFF_MAX_SECONDS = 16.0
 
+# The server caps search results at this many assets.
+SEARCH_RESULT_LIMIT = 10_000
+
 
 def _backoff_delay(attempt: int) -> float:
     """Exponential backoff with jitter: ~1s, 2s, 4s, 8s, 16s."""
@@ -244,7 +249,13 @@ async def POST(
             return r
 
     async def paginated[T: APIResponse](
-        self, path: str, /, *, type: type[T], headers: t.Mapping[str, str] = {}
+        self,
+        path: str,
+        /,
+        *,
+        type: type[T],
+        headers: t.Mapping[str, str] = {},
+        seek: bool = False,
     ) -> t.AsyncGenerator[T, None]:
         """
         Iterate over "data" items in any paged resource.
@@ -253,27 +264,62 @@ async def paginated[T: APIResponse](
             path: the resource endpoint, starting with /
             type: the response JSON type (APIResponse)
             headers: arbitrary HTTP headers for this request
+            seek: work around the server's 10k search-result cap. When a window
+                is exhausted at the cap, restart the search narrowed with `mtf`
+                (modified from) set to the last seen modification time. Requires
+                ascending modification order (`;o=+`) and items exposing
+                `href`/`modified` (i.e. Assets). Boundary assets may repeat
+                (`mtf` is inclusive, minute precision), so assets are deduped per href.
         """
-        page_url: str | None = path
-
-        while page_url:
-            full_results = await self.GET(page_url, headers=headers)
-            full_results = full_results.json()
-
-            # Some first pages are different
-            page: t.Mapping[str, t.Any] = full_results.get("assets", full_results)
-            data = page.get("data", [])
-
-            if len(data) == 0:
+        last_modified: str | None = None
+        seen_hrefs: set[str] = set()
+
+        while True:
+            url = path
+            if last_modified is not None:
+                joiner = "&" if "?" in path else "?"
+                url = f"{path}{joiner}q=mtf%3A{quote(last_modified)}"
+
+            raw = 0
+            yielded = 0
+            page_url: str | None = url
+            while page_url:
+                full_results = await self.GET(page_url, headers=headers)
+                full_results = full_results.json()
+
+                # Some first pages are different
+                page: t.Mapping[str, t.Any] = full_results.get("assets", full_results)
+                data = page.get("data", [])
+
+                if len(data) == 0:
+                    break
+                for d in data:
+                    raw += 1
+                    item = type.model_validate(d)
+                    if seek:
+                        modified = getattr(item, "modified", None)
+                        if modified is not None:
+                            last_modified = modified.astimezone(UTC).strftime(
+                                "%Y-%m-%dT%H:%M:%SZ"
+                            )
+                        href: str = item.href  # type: ignore[attr-defined]
+                        if href in seen_hrefs:
+                            continue
+                        seen_hrefs.add(href)
+                    yield item
+                    yielded += 1
+
+                paging = page.get("paging", {})
+                if paging:
+                    page_url = paging.get("next")
+                else:
+                    page_url = None
+
+            # Stop when a window made no progress (all results were duplicates,
+            # e.g. >10k assets share the boundary timestamp) or when the window
+            # was not capped by the server, meaning the results are exhausted.
+            if not seek or yielded == 0 or raw < SEARCH_RESULT_LIMIT:
                 break
-            for d in data:
-                yield type.model_validate(d)
-
-            paging = page.get("paging", {})
-            if paging:
-                page_url = paging.get("next")
-            else:
-                page_url = None
 
     async def retrying(
         self, path: str, *, retries: int | None = None, delay: float | None = None

src/pyfwapi/tenant.py

@@ -105,11 +105,9 @@ async def match_assets(
                 raise CollectionNotSearchable("Collection '{a}' has no searchURL")
 
             qval = quote(str(query).strip())
-            if qval != "":
-                qval = f"?q={qval}"
-            q = f";o=+{qval}"  # order by oldest modified
+            q = ";o=+" + (f"?q={qval}" if qval else "")  # oldest modified first
             query_url = search_base_url.replace(FOTOWARE_QUERY_PLACEHOLDER, q)
-            async for asset in self.api.paginated(query_url, type=Asset):
+            async for asset in self.api.paginated(query_url, type=Asset, seek=True):
                 yield asset
 
     # MARK: Previews, renditions

tests/test_apiconnection.py

@@ -3,7 +3,8 @@
 import httpxyz as httpx
 import pytest
 
-from pyfwapi.apiconnection import APIConnection
+from pyfwapi.apiconnection import SEARCH_RESULT_LIMIT, APIConnection
+from pyfwapi.model.asset import Asset
 
 
 class TestAPIConnection:
@@ -214,3 +215,132 @@ async def test_post_success(self, api_conn):
             json={"foo": "bar"},
         )
         assert resp.status_code == 201
+
+
+class TestPaginatedSeek:
+    """Seek pagination transparently restarts a search past the 10k cap."""
+
+    LIMIT = SEARCH_RESULT_LIMIT
+
+    @pytest.fixture
+    def mock_client_cls(self):
+        with patch("pyfwapi.apiconnection.AsyncOAuth2Client") as MockClient:
+            instance = MockClient.return_value
+            instance.fetch_token = AsyncMock()
+            instance.aclose = AsyncMock()
+            instance.token = {"access_token": "abc"}
+            yield MockClient, instance
+
+    @pytest.fixture
+    def api_conn(self, mock_client_cls):
+        conn = APIConnection(
+            "https://test.fotoware.cloud/",
+            client_id="test_id",
+            client_secret="test_secret",
+        )
+        conn.GET = AsyncMock()
+        return conn
+
+    @staticmethod
+    def asset_json(i: int, minute: int) -> dict:
+        return {
+            "href": f"/fotoweb/archives/1/asset{i}",
+            "modified": f"2024-01-01T12:{minute:02d}:00Z",
+            "physicalFileId": str(i),
+            "linkstance": "x",
+            "filename": f"asset{i}.jpg",
+            "filesize": 1,
+            "doctype": "image",
+            "created": None,
+            "archiveId": 1,
+            "archiveHREF": "/fotoweb/archives/1",
+            "builtinFields": [],
+            "metadata": {},
+            "previews": None,
+            "previewToken": "x",
+            "renditions": None,
+            "quickRenditions": None,
+        }
+
+    def page(self, assets: list[dict]) -> httpx.Response:
+        return httpx.Response(
+            200,
+            json={"data": assets, "paging": {}},
+            request=httpx.Request("GET", "https://test.fotoware.cloud/search"),
+        )
+
+    @pytest.mark.asyncio
+    async def test_paginated_single_window(self, api_conn):
+        api_conn.GET.return_value = self.page([self.asset_json(i, 0) for i in range(3)])
+
+        assets = [a async for a in api_conn.paginated("/search", type=Asset)]
+
+        assert [a.href for a in assets] == [
+            f"/fotoweb/archives/1/asset{i}" for i in range(3)
+        ]
+        api_conn.GET.assert_awaited_once_with("/search", headers={})
+
+    @pytest.mark.asyncio
+    async def test_paginated_seek_requeries_with_mtf(self, api_conn):
+        first = [self.asset_json(i, 1) for i in range(self.LIMIT)]
+        second = [self.asset_json(i, 2) for i in range(self.LIMIT, self.LIMIT + 5)]
+        api_conn.GET.side_effect = [self.page(first), self.page(second)]
+
+        assets = [
+            a async for a in api_conn.paginated("/search;o=+", type=Asset, seek=True)
+        ]
+
+        assert len(assets) == self.LIMIT + 5
+        assert api_conn.GET.await_count == 2
+        second_url = api_conn.GET.await_args_list[1].args[0]
+        assert second_url == "/search;o=+?q=mtf%3A2024-01-01T12%3A01%3A00Z"
+
+    @pytest.mark.asyncio
+    async def test_paginated_seek_appends_to_existing_query(self, api_conn):
+        first = [self.asset_json(i, 1) for i in range(self.LIMIT)]
+        api_conn.GET.side_effect = [self.page(first), self.page([])]
+
+        assets = [
+            a
+            async for a in api_conn.paginated(
+                "/search;o=+?q=fn%3A%2A.jpg", type=Asset, seek=True
+            )
+        ]
+
+        assert len(assets) == self.LIMIT
+        second_url = api_conn.GET.await_args_list[1].args[0]
+        assert (
+            second_url == "/search;o=+?q=fn%3A%2A.jpg&q=mtf%3A2024-01-01T12%3A01%3A00Z"
+        )
+
+    @pytest.mark.asyncio
+    async def test_paginated_seek_dedupes_boundary_assets(self, api_conn):
+        # all assets share the same minute: window 2 repeats the boundary asset
+        first = [self.asset_json(i, 1) for i in range(self.LIMIT)]
+        second = [first[-1]] + [
+            self.asset_json(i, 1) for i in range(self.LIMIT, self.LIMIT + 5)
+        ]
+        api_conn.GET.side_effect = [self.page(first), self.page(second)]
+
+        assets = [
+            a async for a in api_conn.paginated("/search;o=+", type=Asset, seek=True)
+        ]
+
+        hrefs = [a.href for a in assets]
+        assert len(assets) == self.LIMIT + 5
+        assert len(hrefs) == len(set(hrefs))
+
+    @pytest.mark.asyncio
+    async def test_paginated_seek_all_duplicate_window_breaks_loop(self, api_conn):
+        # >10k assets share the same timestamp: every follow-up window returns
+        # only already-seen assets; the loop must terminate instead of spinning.
+        first = [self.asset_json(i, 1) for i in range(self.LIMIT)]
+        dupes = first[:100]
+        api_conn.GET.side_effect = [self.page(first), self.page(dupes)]
+
+        assets = [
+            a async for a in api_conn.paginated("/search;o=+", type=Asset, seek=True)
+        ]
+
+        assert len(assets) == self.LIMIT
+        assert api_conn.GET.await_count == 2

tests/test_tenant.py

@@ -1,9 +1,11 @@
+from datetime import datetime
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from httpxyz import Response
 
 from pyfwapi.apiconnection import APIConnection
+from pyfwapi.model.asset import Asset
 from pyfwapi.model.collection import Collection
 from pyfwapi.model.instance_info import FullAPIDescriptor
 from pyfwapi.tenant import Tenant
@@ -125,3 +127,59 @@ async def mock_paginated(*args, **kwargs):
         mock_conn.GET.assert_awaited_once_with("/fotoweb/archives/123")
         assert isinstance(archive, Collection)
         assert archive.name == "Test Archive"
+
+
+class TestMatchAssets:
+    @pytest.fixture
+    def mock_conn(self):
+        conn = MagicMock(spec=APIConnection)
+        return conn
+
+    @pytest.fixture
+    def tenant(self, mock_conn):
+        return Tenant(connection=mock_conn)
+
+    @pytest.fixture
+    def archive(self):
+        return Collection.model_construct(
+            id="1",
+            name="Archive 1",
+            href="/fotoweb/archives/1",
+            data="/fotoweb/archives/1/data",
+            type="archive",
+            searchURL="/search{?q}",
+            originalURL="/original",
+            isSearchable=True,
+            permissions=[],
+            canMoveTo=False,
+            canUploadTo=False,
+            description="",
+        )
+
+    @pytest.mark.asyncio
+    async def test_match_assets_seeks_past_10k_limit(self, tenant, mock_conn, archive):
+        """match_assets must order by oldest modified and enable seek pagination,
+        which transparently handles the server's 10k search-result cap."""
+        assets = [
+            Asset.model_construct(
+                href=f"/fotoweb/archives/1/asset{i}",
+                modified=datetime(2024, 1, 1, 12, i),
+            )
+            for i in range(3)
+        ]
+
+        async def gen():
+            for a in assets:
+                yield a
+
+        mock_conn.paginated = MagicMock(return_value=gen())
+
+        result = [
+            a async for a in tenant.match_assets("fn:*.jpg", in_archives=[archive])
+        ]
+
+        assert result == assets
+        mock_conn.paginated.assert_called_once()
+        url = mock_conn.paginated.call_args.args[0]
+        assert url == "/search;o=+?q=fn%3A%2A.jpg"
+        assert mock_conn.paginated.call_args.kwargs["seek"] is True