From 7fc6b3288f5a79ef3eecf3cdb34c96ee1cc4cabf Mon Sep 17 00:00:00 2001 From: Redmer Kronemeijer <12477216+redmer@users.noreply.github.com> Date: Fri, 25 Sep 2026 13:43:54 +0200 Subject: [PATCH 1/3] feat: prevent broken pagination in collection with over 10k assets --- src/pyfwapi/tenant.py | 36 +++++++++++++++++++++++++++++------- 1 file changed, 29 insertions(+), 7 deletions(-) diff --git a/src/pyfwapi/tenant.py b/src/pyfwapi/tenant.py index 32195e1..7ae7c0b 100644 --- a/src/pyfwapi/tenant.py +++ b/src/pyfwapi/tenant.py @@ -104,13 +104,35 @@ async def match_assets( pyfwapiLog.error(f"Collection '{a}' cannot be searched") 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 - query_url = search_base_url.replace(FOTOWARE_QUERY_PLACEHOLDER, q) - async for asset in self.api.paginated(query_url, type=Asset): - yield asset + base_query = str(query).strip() + last_modified: str | None = None + seen = 0 + while True: + # FotoWare caps search results at 10k assets; once we hit the cap, + # restart the search with `mtf` (modified from) after the last seen + # modification time, keeping the ascending modified order. + if last_modified is None: + effective = base_query + else: + boundary = f"mtf:{quote(last_modified)}" + effective = ( + f"{base_query} AND ( {boundary} )" if base_query else boundary + ) + q = ";o=+" + (f"?q={quote(effective)}" if effective else "") + query_url = search_base_url.replace(FOTOWARE_QUERY_PLACEHOLDER, q) + + yielded = 0 + async for asset in self.api.paginated(query_url, type=Asset): + if asset.modified is not None: + last_modified = asset.modified.isoformat( + sep="T", timespec="minutes" + ) + yield asset + yielded += 1 + seen += 1 + if yielded == 0 or seen < 10_000: + break + seen = 0 # MARK: Previews, renditions async def get_preview( From d05c9b4f848df48df3db7b7816f6b16897e3b78a Mon Sep 17 00:00:00 2001 From: Redmer Kronemeijer <12477216+redmer@users.noreply.github.com> Date: Fri, 25 Sep 2026 14:31:42 +0200 Subject: [PATCH 2/3] refactor: on pagination --- src/pyfwapi/apiconnection.py | 86 +++++++++++++++++------ src/pyfwapi/tenant.py | 34 ++------- tests/test_apiconnection.py | 130 ++++++++++++++++++++++++++++++++++- tests/test_tenant.py | 58 ++++++++++++++++ 4 files changed, 258 insertions(+), 50 deletions(-) diff --git a/src/pyfwapi/apiconnection.py b/src/pyfwapi/apiconnection.py index 702b1aa..fc10497 100644 --- a/src/pyfwapi/apiconnection.py +++ b/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 diff --git a/src/pyfwapi/tenant.py b/src/pyfwapi/tenant.py index 7ae7c0b..e1e5bd5 100644 --- a/src/pyfwapi/tenant.py +++ b/src/pyfwapi/tenant.py @@ -104,35 +104,11 @@ async def match_assets( pyfwapiLog.error(f"Collection '{a}' cannot be searched") raise CollectionNotSearchable("Collection '{a}' has no searchURL") - base_query = str(query).strip() - last_modified: str | None = None - seen = 0 - while True: - # FotoWare caps search results at 10k assets; once we hit the cap, - # restart the search with `mtf` (modified from) after the last seen - # modification time, keeping the ascending modified order. - if last_modified is None: - effective = base_query - else: - boundary = f"mtf:{quote(last_modified)}" - effective = ( - f"{base_query} AND ( {boundary} )" if base_query else boundary - ) - q = ";o=+" + (f"?q={quote(effective)}" if effective else "") - query_url = search_base_url.replace(FOTOWARE_QUERY_PLACEHOLDER, q) - - yielded = 0 - async for asset in self.api.paginated(query_url, type=Asset): - if asset.modified is not None: - last_modified = asset.modified.isoformat( - sep="T", timespec="minutes" - ) - yield asset - yielded += 1 - seen += 1 - if yielded == 0 or seen < 10_000: - break - seen = 0 + qval = quote(str(query).strip()) + 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, seek=True): + yield asset # MARK: Previews, renditions async def get_preview( diff --git a/tests/test_apiconnection.py b/tests/test_apiconnection.py index 167d28e..09f82cf 100644 --- a/tests/test_apiconnection.py +++ b/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,130 @@ 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 diff --git a/tests/test_tenant.py b/tests/test_tenant.py index 70043dc..185d448 100644 --- a/tests/test_tenant.py +++ b/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 From def4fbd27bc820bea69b31bf6265d95b829b84cc Mon Sep 17 00:00:00 2001 From: Redmer Kronemeijer <12477216+redmer@users.noreply.github.com> Date: Fri, 25 Sep 2026 14:43:17 +0200 Subject: [PATCH 3/3] test: fix --- tests/test_apiconnection.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_apiconnection.py b/tests/test_apiconnection.py index 09f82cf..b83ab71 100644 --- a/tests/test_apiconnection.py +++ b/tests/test_apiconnection.py @@ -309,7 +309,9 @@ async def test_paginated_seek_appends_to_existing_query(self, api_conn): 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" + 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):