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 32195e1..e1e5bd5 100644 --- a/src/pyfwapi/tenant.py +++ b/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 diff --git a/tests/test_apiconnection.py b/tests/test_apiconnection.py index 167d28e..b83ab71 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,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 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