From 33ec6026f0729bdfbad99c47a464b7e363d56da0 Mon Sep 17 00:00:00 2001 From: 0xSwego <0xSwego@gmail.com> Date: Tue, 4 Aug 2026 22:13:50 +0100 Subject: [PATCH 1/5] feat: add native async STAC resolver --- README.md | 19 +++ dclimate_client_py/__init__.py | 4 + dclimate_client_py/dclimate_client.py | 5 +- dclimate_client_py/stac_server.py | 221 ++++++++++++++++++++++---- pyproject.toml | 2 +- tests/conftest.py | 17 +- tests/test_stac_server_async.py | 176 ++++++++++++++++++++ uv.lock | 12 +- 8 files changed, 411 insertions(+), 45 deletions(-) create mode 100644 tests/test_stac_server_async.py diff --git a/README.md b/README.md index dd8ff99..3460d4b 100644 --- a/README.md +++ b/README.md @@ -141,6 +141,25 @@ for collection_id, info in datasets.items(): ) print(f" Dataset types: {', '.join(info['types'])}") +# Resolve a CID directly without blocking the event loop. Calls made without +# an injected client reuse a pooled httpx.AsyncClient for the current loop. +from dclimate_client_py import ( + aclose_stac_server_client, + aresolve_cid_from_stac_server, +) + +async def resolve_cid(): + try: + resolved = await aresolve_cid_from_stac_server( + collection="ecmwf_aifs", + dataset="temperature_forecast", + variant="single", + ) + print(resolved.cid) + finally: + # Call once when an application event loop shuts down. + await aclose_stac_server_client() + ``` ## Siren API usage diff --git a/dclimate_client_py/__init__.py b/dclimate_client_py/__init__.py index ed2dd9a..20eb826 100644 --- a/dclimate_client_py/__init__.py +++ b/dclimate_client_py/__init__.py @@ -14,6 +14,8 @@ ) from .stac_server import ( ResolvedDataset, + aclose_stac_server_client, + aresolve_cid_from_stac_server, resolve_cid_from_stac_server, list_available_datasets_from_stac_server, STAC_SERVER_URL, @@ -76,6 +78,8 @@ def __dir__() -> list[str]: "load_stac_catalog", "list_available_datasets", "ResolvedDataset", + "aclose_stac_server_client", + "aresolve_cid_from_stac_server", "resolve_cid_from_stac_server", "list_available_datasets_from_stac_server", "STAC_SERVER_URL", diff --git a/dclimate_client_py/dclimate_client.py b/dclimate_client_py/dclimate_client.py index 909f608..3100688 100644 --- a/dclimate_client_py/dclimate_client.py +++ b/dclimate_client_py/dclimate_client.py @@ -25,7 +25,7 @@ from .dclimate_zarr_errors import InvalidSelectionError from .stac_server import ( ResolvedDataset, - resolve_cid_from_stac_server, + aresolve_cid_from_stac_server, list_available_datasets_from_stac_server, ) from .siren import SirenClient @@ -379,8 +379,7 @@ async def load_dataset( # Try STAC server first (faster, avoids loading IPFS catalog) if self._stac_server_url: try: - resolved = await asyncio.to_thread( - resolve_cid_from_stac_server, + resolved = await aresolve_cid_from_stac_server( collection=resolved_collection, dataset=dataset, variant=variant, diff --git a/dclimate_client_py/stac_server.py b/dclimate_client_py/stac_server.py index 5a66995..b65971d 100644 --- a/dclimate_client_py/stac_server.py +++ b/dclimate_client_py/stac_server.py @@ -5,7 +5,9 @@ which is faster than traversing the IPFS-hosted catalog structure. """ -from collections.abc import Iterator +import asyncio +import weakref +from collections.abc import AsyncIterator, Iterator from json import dumps from threading import Lock from typing import Any, Dict, Iterable, NamedTuple, Optional, Set @@ -21,6 +23,10 @@ _MAX_SEARCH_PAGES = 50 _HTTP_CLIENT: httpx.Client | None = None _HTTP_CLIENT_LOCK = Lock() +_ASYNC_HTTP_CLIENTS: weakref.WeakKeyDictionary[ + asyncio.AbstractEventLoop, httpx.AsyncClient +] = weakref.WeakKeyDictionary() +_ASYNC_HTTP_CLIENT_LOCK = Lock() def _client() -> httpx.Client: @@ -33,6 +39,26 @@ def _client() -> httpx.Client: return _HTTP_CLIENT +def _async_client() -> httpx.AsyncClient: + """Return the pooled async STAC client for the current event loop.""" + loop = asyncio.get_running_loop() + with _ASYNC_HTTP_CLIENT_LOCK: + client = _ASYNC_HTTP_CLIENTS.get(loop) + if client is None or client.is_closed: + client = httpx.AsyncClient(timeout=30, follow_redirects=True) + _ASYNC_HTTP_CLIENTS[loop] = client + return client + + +async def aclose_stac_server_client() -> None: + """Close the pooled async STAC client owned by the current event loop.""" + loop = asyncio.get_running_loop() + with _ASYNC_HTTP_CLIENT_LOCK: + client = _ASYNC_HTTP_CLIENTS.pop(loop, None) + if client is not None: + await client.aclose() + + class ResolvedDataset(NamedTuple): cid: str variant: str @@ -238,51 +264,101 @@ def _search_pages( ) -def resolve_cid_from_stac_server( - collection: str, - dataset: str, - variant: Optional[str] = None, - server_url: str = STAC_SERVER_URL, -) -> ResolvedDataset: - """ - Resolve dataset CID via STAC server /search API. +async def _asearch_pages( + server_url: str, + body: Dict[str, Any], + timeout: int, + client: Optional[httpx.AsyncClient] = None, +) -> AsyncIterator[Dict[str, Any]]: + """Yield bounded STAC search pages without blocking the event loop.""" + http_client = client or _async_client() + url = f"{server_url.rstrip('/')}/search" + method = "POST" + request_body: Optional[Dict[str, Any]] = body + request_headers: Dict[str, str] = {} + seen: Set[tuple[str, str, str, str]] = set() - Changed in 0.6: returns ResolvedDataset; variant='' is treated as an - explicit (unresolvable) variant rather than no-variant. + for _ in range(_MAX_SEARCH_PAGES): + page_key = ( + method, + url, + dumps(request_body, sort_keys=True, default=str), + dumps(request_headers, sort_keys=True, default=str), + ) + if page_key in seen: + return + seen.add(page_key) - Uses the same API format as the frontend (POST /search with collections filter). + if method == "POST": + request_kwargs: Dict[str, Any] = { + "json": request_body, + "timeout": timeout, + } + if request_headers: + request_kwargs["headers"] = request_headers + response = await http_client.post(url, **request_kwargs) + else: + request_kwargs = {"params": request_body or None, "timeout": timeout} + if request_headers: + request_kwargs["headers"] = request_headers + response = await http_client.get(url, **request_kwargs) + response.raise_for_status() + page = response.json() + yield page - Args: - collection: Collection ID (e.g., 'ecmwf_aifs', 'ecmwf_era5') - dataset: Dataset name (e.g., 'temperature', 'precipitation') - variant: Optional variant name (e.g., 'ensemble', 'deterministic') - server_url: STAC server base URL + if not (page.get("features") or []): + return + next_link = next( + ( + link + for link in page.get("links", []) or [] + if link.get("rel") == "next" and link.get("href") + ), + None, + ) + if next_link is None: + return - Returns: - ResolvedDataset: The IPFS CID and selected variant + url = urljoin(url, next_link["href"]) + method = str(next_link.get("method", "GET")).upper() + if method not in {"GET", "POST"}: + return + linked_headers = next_link.get("headers") + request_headers = linked_headers if isinstance(linked_headers, dict) else {} + linked_body = next_link.get("body") + if isinstance(linked_body, dict): + if next_link.get("merge"): + request_body = {**body, **linked_body} + else: + request_body = linked_body + elif next_link.get("merge"): + request_body = dict(body) + else: + request_body = None + else: + raise ValueError( + f"STAC search reached its page limit of {_MAX_SEARCH_PAGES} " + "while another next link was present" + ) - Raises: - ValueError: If dataset or variant is not found - httpx.HTTPError: If the server request fails - """ - # Search by collection - body = { - "limit": 100, - "collections": [collection], - } + +def _resolve_dataset_from_features( + collection: str, + dataset: str, + variant: Optional[str], + features: Iterable[Dict[str, Any]], +) -> ResolvedDataset: + """Resolve a dataset from STAC search features shared by both clients.""" + feature_list = list(features) # An item with no variant segment/property is what the listing API # reports as the "default" variant — keep resolve symmetric with list. def _effective_variant(feature: Dict[str, Any]) -> str: return _feature_variant(feature, collection, dataset) or "default" - features: list[Dict[str, Any]] = [] - for page in _search_pages(server_url, body, timeout=10): - features.extend(page.get("features", []) or []) - known_datasets = { dataset_id - for feature in features + for feature in feature_list if isinstance(feature, dict) for dataset_id in [(feature.get("properties") or {}).get("dclimate:dataset_id")] if isinstance(dataset_id, str) and dataset_id @@ -292,7 +368,7 @@ def _effective_variant(feature: Dict[str, Any]) -> str: # as ``precip`` and a known hyphenated dataset ``precip-daily``. matches = [ feature - for feature in features + for feature in feature_list if _feature_matches_dataset( feature, collection, dataset, known_datasets=known_datasets ) @@ -333,6 +409,83 @@ def _effective_variant(feature: Dict[str, Any]) -> str: raise ValueError(f"Item '{item['id']}' has no data asset") +def resolve_cid_from_stac_server( + collection: str, + dataset: str, + variant: Optional[str] = None, + server_url: str = STAC_SERVER_URL, +) -> ResolvedDataset: + """ + Resolve dataset CID via STAC server /search API. + + Changed in 0.6: returns ResolvedDataset; variant='' is treated as an + explicit (unresolvable) variant rather than no-variant. + + Uses the same API format as the frontend (POST /search with collections filter). + + Args: + collection: Collection ID (e.g., 'ecmwf_aifs', 'ecmwf_era5') + dataset: Dataset name (e.g., 'temperature', 'precipitation') + variant: Optional variant name (e.g., 'ensemble', 'deterministic') + server_url: STAC server base URL + + Returns: + ResolvedDataset: The IPFS CID and selected variant + + Raises: + ValueError: If dataset or variant is not found + httpx.HTTPError: If the server request fails + """ + body = { + "limit": 100, + "collections": [collection], + } + features = [ + feature + for page in _search_pages(server_url, body, timeout=10) + for feature in page.get("features", []) or [] + ] + return _resolve_dataset_from_features(collection, dataset, variant, features) + + +async def aresolve_cid_from_stac_server( + collection: str, + dataset: str, + variant: Optional[str] = None, + server_url: str = STAC_SERVER_URL, + *, + client: Optional[httpx.AsyncClient] = None, +) -> ResolvedDataset: + """Resolve a dataset CID natively asynchronously via the STAC API. + + When ``client`` is omitted, calls reuse a pooled ``httpx.AsyncClient`` + scoped to the current event loop. Applications may inject their own client + when they need custom transport settings or lifecycle control. + + Args: + collection: Collection ID (e.g., 'ecmwf_aifs', 'ecmwf_era5') + dataset: Dataset name (e.g., 'temperature', 'precipitation') + variant: Optional variant name (e.g., 'ensemble', 'deterministic') + server_url: STAC server base URL + client: Optional caller-owned pooled async HTTP client + + Returns: + ResolvedDataset: The IPFS CID and selected variant + + Raises: + ValueError: If dataset or variant is not found + httpx.HTTPError: If the server request fails + """ + body = { + "limit": 100, + "collections": [collection], + } + features: list[Dict[str, Any]] = [] + async for page in _asearch_pages(server_url, body, timeout=10, client=client): + features.extend(page.get("features", []) or []) + return _resolve_dataset_from_features(collection, dataset, variant, features) + + def _strip_ipfs_scheme(cid: Optional[str]) -> Optional[str]: if not cid: return None diff --git a/pyproject.toml b/pyproject.toml index 0cf058c..609b65f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "pdm.backend" [project] name = "dclimate-client-py" -version = "0.6.1" # Set a static version or handle it in versioning strategy +version = "0.7.0" # Set a static version or handle it in versioning strategy description = "Python client library for accessing dClimate weather and climate data" readme = "README.md" license = {text = "MIT"} diff --git a/tests/conftest.py b/tests/conftest.py index c91e9de..c025b10 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,3 +1,4 @@ +import asyncio import datetime import itertools import pathlib @@ -15,8 +16,9 @@ @pytest.fixture def install_httpx_mock(monkeypatch): - """Inject a pooled MockTransport client through a module's client accessor.""" + """Inject pooled sync and async clients through a module's accessors.""" clients: list[httpx.Client] = [] + async_clients: list[httpx.AsyncClient] = [] def install(module, handler): client = httpx.Client( @@ -24,14 +26,27 @@ def install(module, handler): timeout=30, follow_redirects=True, ) + + async def async_handler(request: httpx.Request) -> httpx.Response: + return await asyncio.to_thread(handler, request) + + async_client = httpx.AsyncClient( + transport=httpx.MockTransport(async_handler), + timeout=30, + follow_redirects=True, + ) clients.append(client) + async_clients.append(async_client) monkeypatch.setattr(module, "_client", lambda: client) + monkeypatch.setattr(module, "_async_client", lambda: async_client) return client yield install for client in clients: client.close() + for async_client in async_clients: + asyncio.run(async_client.aclose()) def pytest_addoption(parser): diff --git a/tests/test_stac_server_async.py b/tests/test_stac_server_async.py new file mode 100644 index 0000000..41f46ff --- /dev/null +++ b/tests/test_stac_server_async.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +from unittest.mock import AsyncMock + +import httpx +import pytest +import xarray as xr + +from dclimate_client_py import dclimate_client, stac_server +from dclimate_client_py.stac_server import ResolvedDataset + + +COLLECTION = "example_collection" +DATASET = "temperature_mean" + + +def _feature(dataset: str, variant: str, cid: str) -> dict: + return { + "id": f"{COLLECTION}-{dataset}-{variant}", + "collection": COLLECTION, + "properties": { + "dclimate:dataset_id": dataset, + "dclimate:variant": variant, + }, + "assets": {"data": {"href": f"ipfs://{cid}"}}, + } + + +@pytest.mark.asyncio +async def test_async_resolver_follows_pagination_with_injected_client( + monkeypatch: pytest.MonkeyPatch, +) -> None: + requests: list[tuple[str, str]] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + requests.append((request.method, str(request.url))) + if request.url.params.get("page") == "2": + payload = {"features": [_feature(DATASET, "default", "bafy-target")]} + else: + payload = { + "features": [_feature("other_dataset", "default", "bafy-other")], + "links": [ + { + "rel": "next", + "href": "/search?page=2", + "method": "GET", + } + ], + } + return httpx.Response(200, json=payload, request=request) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + monkeypatch.setattr( + stac_server, + "_async_client", + lambda: (_ for _ in ()).throw(AssertionError("injected client ignored")), + ) + + resolved = await stac_server.aresolve_cid_from_stac_server( + collection=COLLECTION, + dataset=DATASET, + variant="default", + server_url="https://stac.example", + client=client, + ) + + assert resolved == ResolvedDataset(cid="bafy-target", variant="default") + assert requests == [ + ("POST", "https://stac.example/search"), + ("GET", "https://stac.example/search?page=2"), + ] + + +@pytest.mark.asyncio +async def test_async_resolver_reuses_and_closes_loop_local_client( + monkeypatch: pytest.MonkeyPatch, +) -> None: + original_async_client = httpx.AsyncClient + clients: list[httpx.AsyncClient] = [] + request_count = 0 + + async def handler(request: httpx.Request) -> httpx.Response: + nonlocal request_count + request_count += 1 + return httpx.Response( + 200, + json={"features": [_feature(DATASET, "default", "bafy-pooled")]}, + request=request, + ) + + def client_factory(*args, **kwargs): # type: ignore[no-untyped-def] + client = original_async_client( + *args, + **kwargs, + transport=httpx.MockTransport(handler), + ) + clients.append(client) + return client + + monkeypatch.setattr(stac_server.httpx, "AsyncClient", client_factory) + stac_server._ASYNC_HTTP_CLIENTS.clear() + + first = await stac_server.aresolve_cid_from_stac_server( + collection=COLLECTION, + dataset=DATASET, + server_url="https://stac.example", + ) + second = await stac_server.aresolve_cid_from_stac_server( + collection=COLLECTION, + dataset=DATASET, + server_url="https://stac.example", + ) + + assert first == second == ResolvedDataset("bafy-pooled", "default") + assert len(clients) == 1 + assert request_count == 2 + assert clients[0].is_closed is False + + await stac_server.aclose_stac_server_client() + + assert clients[0].is_closed is True + assert not stac_server._ASYNC_HTTP_CLIENTS + + +@pytest.mark.asyncio +async def test_async_resolver_matches_sync_selection_rules() -> None: + features = [ + _feature(DATASET, "latest", "bafy-latest"), + _feature(DATASET, "final", "bafy-final"), + _feature(DATASET, "default", "bafy-default"), + ] + + async def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"features": features}, request=request) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + resolved = await stac_server.aresolve_cid_from_stac_server( + collection=COLLECTION, + dataset=DATASET, + server_url="https://stac.example", + client=client, + ) + + assert resolved == ResolvedDataset("bafy-default", "default") + + +@pytest.mark.asyncio +async def test_high_level_client_uses_native_async_resolver( + monkeypatch: pytest.MonkeyPatch, +) -> None: + resolver = AsyncMock(return_value=ResolvedDataset("bafy-native", "default")) + + async def load_from_ipfs(**kwargs): # type: ignore[no-untyped-def] + assert kwargs["ipfs_cid"] == "bafy-native" + return xr.Dataset({"temperature": ("time", [21.0])}, coords={"time": [0]}) + + monkeypatch.setattr(dclimate_client, "aresolve_cid_from_stac_server", resolver) + monkeypatch.setattr(dclimate_client, "_load_dataset_from_ipfs_cid", load_from_ipfs) + client = dclimate_client.dClimateClient(stac_server_url="https://stac.example") + client._kubo_cas = object() + + dataset, metadata = await client.load_dataset( + collection=COLLECTION, + dataset=DATASET, + variant="default", + return_xarray=True, + ) + + resolver.assert_awaited_once_with( + collection=COLLECTION, + dataset=DATASET, + variant="default", + server_url="https://stac.example", + ) + assert dataset["temperature"].values.tolist() == [21.0] + assert metadata["cid"] == "bafy-native" diff --git a/uv.lock b/uv.lock index 58a7bbf..6390360 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = ">=3.12" resolution-markers = [ "python_full_version >= '3.15'", @@ -677,7 +677,7 @@ wheels = [ [[package]] name = "dclimate-client-py" -version = "0.6.0" +version = "0.7.0" source = { editable = "." } dependencies = [ { name = "aiobotocore" }, @@ -730,7 +730,7 @@ requires-dist = [ { name = "opentelemetry-api", specifier = ">=1.30.0" }, { name = "pandas", specifier = ">=2.2.0" }, { name = "pre-commit", marker = "extra == 'dev'", specifier = ">=4.1.0" }, - { name = "py-hamt", specifier = ">=3.5.0" }, + { name = "py-hamt", specifier = ">=3.6.0" }, { name = "pycryptodome", specifier = ">=3.21.0" }, { name = "pystac", specifier = ">=1.10.0" }, { name = "pytest", marker = "extra == 'testing'" }, @@ -1631,7 +1631,7 @@ wheels = [ [[package]] name = "py-hamt" -version = "3.5.0" +version = "3.6.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "dag-cbor" }, @@ -1642,9 +1642,9 @@ dependencies = [ { name = "pycryptodome" }, { name = "zarr" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/ba/ad/42f19faceb109c85f457249b77aa6da95f63b30ff08aa350be32197b251e/py_hamt-3.5.0.tar.gz", hash = "sha256:5f0cc81fd9a360c481bc835163d2acfaa0862cf6544f5f9ea103cc3b1c17bd70", size = 296183, upload-time = "2026-07-21T15:11:40.521Z" } +sdist = { url = "https://files.pythonhosted.org/packages/a6/4a/783c03018f642a8df17b1dbfb46968209c656636dd0e5c1008cb80320609/py_hamt-3.6.0.tar.gz", hash = "sha256:7839aeb9a5ec8fede281521b10afdfc632a119c2db1dd7fd7d2374e9eca0fd09", size = 326698, upload-time = "2026-07-30T12:15:35.036Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/62/e4/c003760c4af897b47e1e18bef8bfa828f279cb9b1fd7ee86ffdf290b0b37/py_hamt-3.5.0-py3-none-any.whl", hash = "sha256:5e06627d3105ff92b3ece3f0dae5ebd9003117ea976a8740919dfec47eeaffb2", size = 70398, upload-time = "2026-07-21T15:11:39.513Z" }, + { url = "https://files.pythonhosted.org/packages/c1/f3/8ea8b9d21e0b36490bff3e3524994d64d3365ef184138f2b568fa0326ed7/py_hamt-3.6.0-py3-none-any.whl", hash = "sha256:9b20fa474b84f5a112ebd61093fbab2613117b0ede8fa4fd5114605110e647da", size = 81309, upload-time = "2026-07-30T12:15:33.917Z" }, ] [[package]] From f9b258c7d0e81d95cb5d87aef109fbda9a56c6d1 Mon Sep 17 00:00:00 2001 From: 0xSwego <0xSwego@gmail.com> Date: Tue, 4 Aug 2026 22:46:07 +0100 Subject: [PATCH 2/5] fix: address async resolver review feedback --- dclimate_client_py/dclimate_client.py | 74 +++++--- dclimate_client_py/stac_server.py | 241 ++++++++++++++++---------- tests/test_review_bugs_client.py | 28 +++ tests/test_review_fu_variant.py | 2 + tests/test_review_perf_async.py | 1 + tests/test_stac_server_async.py | 125 ++++++++++++- 6 files changed, 346 insertions(+), 125 deletions(-) diff --git a/dclimate_client_py/dclimate_client.py b/dclimate_client_py/dclimate_client.py index 3100688..01a57d1 100644 --- a/dclimate_client_py/dclimate_client.py +++ b/dclimate_client_py/dclimate_client.py @@ -39,6 +39,23 @@ DEFAULT_PUBLIC_GATEWAY = "https://ipfs-gateway.dclimate.net" +def _merge_cleanup_error( + current: BaseException | None, + new: BaseException, +) -> BaseException: + """Chain cleanup failures while ensuring cancellation remains dominant.""" + if current is None: + return new + if isinstance(current, asyncio.CancelledError) and not isinstance( + new, asyncio.CancelledError + ): + new.__context__ = current.__context__ + current.__context__ = new + return current + new.__context__ = current + return new + + class dClimateClient: """ Async context manager for loading dClimate datasets from IPFS. @@ -133,6 +150,7 @@ def __init__( self._client_factory = client_factory self._stac_catalog: typing.Optional["pystac.Catalog"] = None self._stac_catalog_lock = asyncio.Lock() + self._stac_http_client: typing.Optional[httpx.AsyncClient] = None self._kubo_cas: typing.Optional[KuboCAS] = None # Note: STAC catalog is loaded lazily (only if STAC server fails) @@ -170,53 +188,52 @@ async def __aenter__(self) -> "dClimateClient": return self async def __aexit__(self, exc_type, exc_val, exc_tb): - """Clean up KuboCAS when exiting async context.""" + """Clean up owned HTTP and KuboCAS resources.""" incoming_cancellation = isinstance(exc_val, asyncio.CancelledError) - siren_error: BaseException | None = None + cleanup_error: BaseException | None = None + + try: + if self._stac_http_client is not None: + await self._stac_http_client.aclose() + except BaseException as error: + cleanup_error = _merge_cleanup_error(cleanup_error, error) + finally: + self._stac_http_client = None + try: if self._siren_client is not None: await self._siren_client.aclose() except BaseException as error: - siren_error = error + cleanup_error = _merge_cleanup_error(cleanup_error, error) try: if self._kubo_cas is not None: await self._kubo_cas.__aexit__(exc_type, exc_val, exc_tb) except BaseException as kubo_error: - if incoming_cancellation and not isinstance( - kubo_error, asyncio.CancelledError - ): - # Preserve cancellation from the context body. An ordinary - # cleanup failure must not replace task cancellation, but - # remains inspectable through the exception context chain. - if siren_error is not None: - kubo_error.__context__ = siren_error - exc_val.__context__ = kubo_error - return False - if siren_error is None: - raise - # Both cleanups failed. Follow the AsyncExitStack convention (the - # later error propagates with the earlier as __context__), except - # that a cancellation always outranks an ordinary error. - if isinstance(siren_error, asyncio.CancelledError) and not isinstance( - kubo_error, asyncio.CancelledError - ): - raise siren_error - kubo_error.__context__ = siren_error - raise + cleanup_error = _merge_cleanup_error(cleanup_error, kubo_error) finally: self._kubo_cas = None - if siren_error is not None: + if cleanup_error is not None: if incoming_cancellation and not isinstance( - siren_error, asyncio.CancelledError + cleanup_error, asyncio.CancelledError ): - exc_val.__context__ = siren_error + # Preserve cancellation from the context body. Ordinary + # cleanup failures remain inspectable through its context. + exc_val.__context__ = cleanup_error else: - raise siren_error + raise cleanup_error return False + def _get_stac_http_client(self) -> httpx.AsyncClient: + """Return the pooled STAC transport owned by this client context.""" + client = self._stac_http_client + if client is None or client.is_closed: + client = httpx.AsyncClient(timeout=30, follow_redirects=True) + self._stac_http_client = client + return client + @staticmethod def _apply_zarr_group_metadata(ds: xr.Dataset, metadata: DatasetMetadata) -> None: loaded_zarr_group = ds.attrs.get("_ipfs_zarr_group") @@ -384,6 +401,7 @@ async def load_dataset( dataset=dataset, variant=variant, server_url=self._stac_server_url, + client=self._get_stac_http_client(), ) except (httpx.HTTPError, ValueError): # Fall back when server lookup fails or returns no usable match. diff --git a/dclimate_client_py/stac_server.py b/dclimate_client_py/stac_server.py index b65971d..5972c10 100644 --- a/dclimate_client_py/stac_server.py +++ b/dclimate_client_py/stac_server.py @@ -11,7 +11,7 @@ from json import dumps from threading import Lock from typing import Any, Dict, Iterable, NamedTuple, Optional, Set -from urllib.parse import urljoin +from urllib.parse import urljoin, urlsplit import httpx @@ -29,6 +29,12 @@ _ASYNC_HTTP_CLIENT_LOCK = Lock() +_SearchBody = Optional[Dict[str, Any]] +_SearchHeaders = Dict[str, str] +_SearchRequest = tuple[str, str, _SearchBody, _SearchHeaders] +_SearchPageKey = tuple[str, str, str, str] + + def _client() -> httpx.Client: """Return the process-wide pooled client used for synchronous STAC calls.""" global _HTTP_CLIENT @@ -180,6 +186,121 @@ def _feature_matches_dataset( return _dataset_id_from_item_id(feature_id, collection, dataset) == dataset +def _search_page_key( + method: str, + url: str, + body: _SearchBody, + headers: _SearchHeaders, +) -> _SearchPageKey: + """Return a stable key used to stop repeated pagination requests.""" + return ( + method, + url, + dumps(body, sort_keys=True, default=str), + dumps(headers, sort_keys=True, default=str), + ) + + +def _search_request_kwargs( + method: str, + body: _SearchBody, + headers: _SearchHeaders, + timeout: int, +) -> Dict[str, Any]: + """Build transport-independent request arguments for a search page.""" + if method == "POST": + request_kwargs: Dict[str, Any] = {"json": body, "timeout": timeout} + else: + request_kwargs = {"params": body or None, "timeout": timeout} + if headers: + request_kwargs["headers"] = headers + return request_kwargs + + +def _url_origin(url: str) -> tuple[str, str, Optional[int]]: + """Return a normalized URL origin for STAC pagination validation.""" + parsed = urlsplit(url) + scheme = parsed.scheme.lower() + hostname = (parsed.hostname or "").rstrip(".").lower() + if not scheme or not hostname: + raise ValueError(f"STAC pagination link is not an absolute URL: {url!r}") + try: + port = parsed.port + except ValueError as error: + raise ValueError( + f"STAC pagination link has an invalid port: {url!r}" + ) from error + if port is None: + port = {"http": 80, "https": 443}.get(scheme) + return scheme, hostname, port + + +def _next_search_request( + server_url: str, + current_url: str, + original_body: Dict[str, Any], + page: Dict[str, Any], +) -> Optional[_SearchRequest]: + """Plan the next same-origin STAC request from a search response page.""" + next_link = next( + ( + link + for link in page.get("links", []) or [] + if link.get("rel") == "next" and link.get("href") + ), + None, + ) + if next_link is None: + return None + + href = next_link["href"] + if not isinstance(href, str): + raise ValueError("STAC pagination link href must be a string") + next_url = urljoin(current_url, href) + parsed_next_url = urlsplit(next_url) + if ( + _url_origin(next_url) != _url_origin(server_url) + or parsed_next_url.username is not None + or parsed_next_url.password is not None + ): + raise ValueError( + "STAC pagination link must use the configured server origin " + f"{_url_origin(server_url)!r}: {next_url!r}" + ) + + method = str(next_link.get("method", "GET")).upper() + if method not in {"GET", "POST"}: + return None + + linked_headers = next_link.get("headers") + # A server-provided next link may carry continuation credentials. Forward + # those headers only on a validated, encrypted connection to the same + # origin; plaintext endpoints still paginate without linked headers. + request_headers = ( + linked_headers + if parsed_next_url.scheme.lower() == "https" + and isinstance(linked_headers, dict) + else {} + ) + + linked_body = next_link.get("body") + if isinstance(linked_body, dict): + # STAC API next-link contract: with "merge": true the linked body + # extends the original request (keeping filters like "collections"); + # otherwise it replaces it wholesale. + request_body = ( + {**original_body, **linked_body} if next_link.get("merge") else linked_body + ) + elif next_link.get("merge"): + # ``merge: true`` without a link body still carries the original + # search filters to the next request. + request_body = dict(original_body) + else: + request_body = None + + return next_url, method, request_body, request_headers + + def _search_pages( server_url: str, body: Dict[str, Any], @@ -188,33 +309,22 @@ def _search_pages( """Yield bounded STAC search pages while following ``rel=next`` links.""" url = f"{server_url.rstrip('/')}/search" method = "POST" - request_body: Optional[Dict[str, Any]] = body - request_headers: Dict[str, str] = {} - seen: Set[tuple[str, str, str, str]] = set() + request_body: _SearchBody = body + request_headers: _SearchHeaders = {} + seen: Set[_SearchPageKey] = set() for _ in range(_MAX_SEARCH_PAGES): - page_key = ( - method, - url, - dumps(request_body, sort_keys=True, default=str), - dumps(request_headers, sort_keys=True, default=str), - ) + page_key = _search_page_key(method, url, request_body, request_headers) if page_key in seen: return seen.add(page_key) + request_kwargs = _search_request_kwargs( + method, request_body, request_headers, timeout + ) if method == "POST": - request_kwargs: Dict[str, Any] = { - "json": request_body, - "timeout": timeout, - } - if request_headers: - request_kwargs["headers"] = request_headers response = _client().post(url, **request_kwargs) else: - request_kwargs = {"params": request_body or None, "timeout": timeout} - if request_headers: - request_kwargs["headers"] = request_headers response = _client().get(url, **request_kwargs) response.raise_for_status() page = response.json() @@ -222,38 +332,10 @@ def _search_pages( if not (page.get("features") or []): return - next_link = next( - ( - link - for link in page.get("links", []) or [] - if link.get("rel") == "next" and link.get("href") - ), - None, - ) - if next_link is None: + next_request = _next_search_request(server_url, url, body, page) + if next_request is None: return - - url = urljoin(url, next_link["href"]) - method = str(next_link.get("method", "GET")).upper() - if method not in {"GET", "POST"}: - return - linked_headers = next_link.get("headers") - request_headers = linked_headers if isinstance(linked_headers, dict) else {} - linked_body = next_link.get("body") - if isinstance(linked_body, dict): - # STAC API next-link contract: with "merge": true the linked - # body extends the original request (keeping filters like - # "collections"); otherwise it replaces it wholesale. - if next_link.get("merge"): - request_body = {**body, **linked_body} - else: - request_body = linked_body - elif next_link.get("merge"): - # ``merge: true`` without a link body still carries the original - # search filters to the next request. - request_body = dict(body) - else: - request_body = None + url, method, request_body, request_headers = next_request else: # Reaching the bound with a valid next link means the result is # incomplete. Surface that explicitly so callers can use their @@ -274,33 +356,22 @@ async def _asearch_pages( http_client = client or _async_client() url = f"{server_url.rstrip('/')}/search" method = "POST" - request_body: Optional[Dict[str, Any]] = body - request_headers: Dict[str, str] = {} - seen: Set[tuple[str, str, str, str]] = set() + request_body: _SearchBody = body + request_headers: _SearchHeaders = {} + seen: Set[_SearchPageKey] = set() for _ in range(_MAX_SEARCH_PAGES): - page_key = ( - method, - url, - dumps(request_body, sort_keys=True, default=str), - dumps(request_headers, sort_keys=True, default=str), - ) + page_key = _search_page_key(method, url, request_body, request_headers) if page_key in seen: return seen.add(page_key) + request_kwargs = _search_request_kwargs( + method, request_body, request_headers, timeout + ) if method == "POST": - request_kwargs: Dict[str, Any] = { - "json": request_body, - "timeout": timeout, - } - if request_headers: - request_kwargs["headers"] = request_headers response = await http_client.post(url, **request_kwargs) else: - request_kwargs = {"params": request_body or None, "timeout": timeout} - if request_headers: - request_kwargs["headers"] = request_headers response = await http_client.get(url, **request_kwargs) response.raise_for_status() page = response.json() @@ -308,33 +379,10 @@ async def _asearch_pages( if not (page.get("features") or []): return - next_link = next( - ( - link - for link in page.get("links", []) or [] - if link.get("rel") == "next" and link.get("href") - ), - None, - ) - if next_link is None: + next_request = _next_search_request(server_url, url, body, page) + if next_request is None: return - - url = urljoin(url, next_link["href"]) - method = str(next_link.get("method", "GET")).upper() - if method not in {"GET", "POST"}: - return - linked_headers = next_link.get("headers") - request_headers = linked_headers if isinstance(linked_headers, dict) else {} - linked_body = next_link.get("body") - if isinstance(linked_body, dict): - if next_link.get("merge"): - request_body = {**body, **linked_body} - else: - request_body = linked_body - elif next_link.get("merge"): - request_body = dict(body) - else: - request_body = None + url, method, request_body, request_headers = next_request else: raise ValueError( f"STAC search reached its page limit of {_MAX_SEARCH_PAGES} " @@ -459,8 +507,9 @@ async def aresolve_cid_from_stac_server( """Resolve a dataset CID natively asynchronously via the STAC API. When ``client`` is omitted, calls reuse a pooled ``httpx.AsyncClient`` - scoped to the current event loop. Applications may inject their own client - when they need custom transport settings or lifecycle control. + scoped to the current event loop. Call ``aclose_stac_server_client`` when + that loop shuts down. Injected clients remain caller-owned and are never + closed by this function. Args: collection: Collection ID (e.g., 'ecmwf_aifs', 'ecmwf_era5') diff --git a/tests/test_review_bugs_client.py b/tests/test_review_bugs_client.py index e553a7a..d694496 100644 --- a/tests/test_review_bugs_client.py +++ b/tests/test_review_bugs_client.py @@ -1,6 +1,7 @@ import asyncio from unittest.mock import AsyncMock +import httpx import pytest from dclimate_client_py.dclimate_client import dClimateClient @@ -33,6 +34,33 @@ async def test_aexit_forwards_with_block_exception_to_kubo(): assert client._kubo_cas is None +@pytest.mark.asyncio +async def test_aexit_closes_owned_stac_client(): + client, kubo_cas, _ = _client_with_mocks() + stac_http_client = AsyncMock(spec=httpx.AsyncClient) + client._stac_http_client = stac_http_client + + await client.__aexit__(None, None, None) + + stac_http_client.aclose.assert_awaited_once() + kubo_cas.__aexit__.assert_awaited_once_with(None, None, None) + assert client._stac_http_client is None + + +@pytest.mark.asyncio +async def test_aexit_still_closes_kubo_when_stac_close_raises(): + client, kubo_cas, _ = _client_with_mocks() + stac_http_client = AsyncMock(spec=httpx.AsyncClient) + stac_http_client.aclose.side_effect = RuntimeError("STAC close failed") + client._stac_http_client = stac_http_client + + with pytest.raises(RuntimeError, match="STAC close failed"): + await client.__aexit__(None, None, None) + + kubo_cas.__aexit__.assert_awaited_once_with(None, None, None) + assert client._stac_http_client is None + + @pytest.mark.asyncio async def test_aexit_dual_failure_propagates_later_error_with_context(): siren_error = RuntimeError("Siren close failed") diff --git a/tests/test_review_fu_variant.py b/tests/test_review_fu_variant.py index 761405d..c4ad118 100644 --- a/tests/test_review_fu_variant.py +++ b/tests/test_review_fu_variant.py @@ -128,6 +128,7 @@ async def test_load_dataset_reports_variant_selected_by_stac_server(monkeypatch) _stub_dataset_loader, ) client = dClimateClient(stac_server_url="https://stac.example") + client._stac_http_client = stac_server._async_client() client._kubo_cas = object() _, metadata = await client.load_dataset( @@ -200,6 +201,7 @@ async def test_explicit_variant_is_preserved_in_loaded_metadata(monkeypatch): _stub_dataset_loader, ) client = dClimateClient(stac_server_url="https://stac.example") + client._stac_http_client = stac_server._async_client() client._kubo_cas = object() _, metadata = await client.load_dataset( diff --git a/tests/test_review_perf_async.py b/tests/test_review_perf_async.py index df38ee0..3ad2af6 100644 --- a/tests/test_review_perf_async.py +++ b/tests/test_review_perf_async.py @@ -39,6 +39,7 @@ async def load_from_ipfs(**kwargs): monkeypatch.setattr(dclimate_client, "_load_dataset_from_ipfs_cid", load_from_ipfs) client = dclimate_client.dClimateClient(stac_server_url="https://stac.invalid") + client._stac_http_client = stac_server._async_client() client._kubo_cas = AsyncMock() loop = asyncio.get_running_loop() diff --git a/tests/test_stac_server_async.py b/tests/test_stac_server_async.py index 41f46ff..4d291b8 100644 --- a/tests/test_stac_server_async.py +++ b/tests/test_stac_server_async.py @@ -26,6 +26,28 @@ def _feature(dataset: str, variant: str, cid: str) -> dict: } +async def _close_all_pooled_clients() -> None: + """Close registry clients before removing the test-owned references.""" + errors: list[BaseException] = [] + for client in list(stac_server._ASYNC_HTTP_CLIENTS.values()): + try: + await client.aclose() + except BaseException as error: + errors.append(error) + stac_server._ASYNC_HTTP_CLIENTS.clear() + if errors: + raise errors[0] + + +@pytest.fixture(autouse=True) +async def close_pooled_clients_between_tests(): + await _close_all_pooled_clients() + try: + yield + finally: + await _close_all_pooled_clients() + + @pytest.mark.asyncio async def test_async_resolver_follows_pagination_with_injected_client( monkeypatch: pytest.MonkeyPatch, @@ -71,6 +93,104 @@ async def handler(request: httpx.Request) -> httpx.Response: ] +@pytest.mark.asyncio +@pytest.mark.parametrize( + "next_href", + [ + "https://attacker.example/collect", + "http://stac.example/search?page=2", + ], +) +async def test_async_resolver_rejects_untrusted_pagination_links( + next_href: str, +) -> None: + requests: list[str] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + requests.append(str(request.url)) + return httpx.Response( + 200, + json={ + "features": [_feature("other_dataset", "default", "bafy-other")], + "links": [{"rel": "next", "href": next_href}], + }, + request=request, + ) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + with pytest.raises(ValueError, match="configured server origin"): + await stac_server.aresolve_cid_from_stac_server( + collection=COLLECTION, + dataset=DATASET, + server_url="https://stac.example", + client=client, + ) + + assert requests == ["https://stac.example/search"] + + +@pytest.mark.asyncio +async def test_async_resolver_drops_linked_headers_on_plaintext_pagination() -> None: + requests: list[httpx.Request] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + if len(requests) == 1: + payload = { + "features": [_feature("other_dataset", "default", "bafy-other")], + "links": [ + { + "rel": "next", + "href": "/search?page=2", + "headers": {"Authorization": "Bearer continuation"}, + } + ], + } + else: + payload = {"features": [_feature(DATASET, "default", "bafy-target")]} + return httpx.Response(200, json=payload, request=request) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + resolved = await stac_server.aresolve_cid_from_stac_server( + collection=COLLECTION, + dataset=DATASET, + server_url="http://stac.example", + client=client, + ) + + assert resolved.cid == "bafy-target" + assert len(requests) == 2 + assert "authorization" not in requests[1].headers + + +def test_sync_resolver_rejects_untrusted_pagination_link( + monkeypatch: pytest.MonkeyPatch, +) -> None: + requests: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(str(request.url)) + return httpx.Response( + 200, + json={ + "features": [_feature("other_dataset", "default", "bafy-other")], + "links": [{"rel": "next", "href": "https://attacker.example/collect"}], + }, + request=request, + ) + + with httpx.Client(transport=httpx.MockTransport(handler)) as client: + monkeypatch.setattr(stac_server, "_client", lambda: client) + with pytest.raises(ValueError, match="configured server origin"): + stac_server.resolve_cid_from_stac_server( + collection=COLLECTION, + dataset=DATASET, + server_url="https://stac.example", + ) + + assert requests == ["https://stac.example/search"] + + @pytest.mark.asyncio async def test_async_resolver_reuses_and_closes_loop_local_client( monkeypatch: pytest.MonkeyPatch, @@ -98,7 +218,6 @@ def client_factory(*args, **kwargs): # type: ignore[no-untyped-def] return client monkeypatch.setattr(stac_server.httpx, "AsyncClient", client_factory) - stac_server._ASYNC_HTTP_CLIENTS.clear() first = await stac_server.aresolve_cid_from_stac_server( collection=COLLECTION, @@ -158,6 +277,9 @@ async def load_from_ipfs(**kwargs): # type: ignore[no-untyped-def] monkeypatch.setattr(dclimate_client, "_load_dataset_from_ipfs_cid", load_from_ipfs) client = dclimate_client.dClimateClient(stac_server_url="https://stac.example") client._kubo_cas = object() + stac_http_client = AsyncMock(spec=httpx.AsyncClient) + stac_http_client.is_closed = False + client._stac_http_client = stac_http_client dataset, metadata = await client.load_dataset( collection=COLLECTION, @@ -171,6 +293,7 @@ async def load_from_ipfs(**kwargs): # type: ignore[no-untyped-def] dataset=DATASET, variant="default", server_url="https://stac.example", + client=stac_http_client, ) assert dataset["temperature"].values.tolist() == [21.0] assert metadata["cid"] == "bafy-native" From 58ee6e8209db8fac7dbef3ac1e18d18d179d097f Mon Sep 17 00:00:00 2001 From: 0xSwego <0xSwego@gmail.com> Date: Tue, 4 Aug 2026 22:54:02 +0100 Subject: [PATCH 3/5] fix: harden STAC request handling --- dclimate_client_py/dclimate_client.py | 2 +- dclimate_client_py/stac_server.py | 18 +++++--- tests/test_stac_server_async.py | 59 +++++++++++++++++++++++++++ 3 files changed, 73 insertions(+), 6 deletions(-) diff --git a/dclimate_client_py/dclimate_client.py b/dclimate_client_py/dclimate_client.py index 01a57d1..230cf6c 100644 --- a/dclimate_client_py/dclimate_client.py +++ b/dclimate_client_py/dclimate_client.py @@ -230,7 +230,7 @@ def _get_stac_http_client(self) -> httpx.AsyncClient: """Return the pooled STAC transport owned by this client context.""" client = self._stac_http_client if client is None or client.is_closed: - client = httpx.AsyncClient(timeout=30, follow_redirects=True) + client = httpx.AsyncClient(timeout=30, follow_redirects=False) self._stac_http_client = client return client diff --git a/dclimate_client_py/stac_server.py b/dclimate_client_py/stac_server.py index 5972c10..8eadb2d 100644 --- a/dclimate_client_py/stac_server.py +++ b/dclimate_client_py/stac_server.py @@ -41,7 +41,7 @@ def _client() -> httpx.Client: if _HTTP_CLIENT is None: with _HTTP_CLIENT_LOCK: if _HTTP_CLIENT is None: - _HTTP_CLIENT = httpx.Client(timeout=30, follow_redirects=True) + _HTTP_CLIENT = httpx.Client(timeout=30, follow_redirects=False) return _HTTP_CLIENT @@ -51,7 +51,7 @@ def _async_client() -> httpx.AsyncClient: with _ASYNC_HTTP_CLIENT_LOCK: client = _ASYNC_HTTP_CLIENTS.get(loop) if client is None or client.is_closed: - client = httpx.AsyncClient(timeout=30, follow_redirects=True) + client = httpx.AsyncClient(timeout=30, follow_redirects=False) _ASYNC_HTTP_CLIENTS[loop] = client return client @@ -209,9 +209,17 @@ def _search_request_kwargs( ) -> Dict[str, Any]: """Build transport-independent request arguments for a search page.""" if method == "POST": - request_kwargs: Dict[str, Any] = {"json": body, "timeout": timeout} + request_kwargs: Dict[str, Any] = { + "json": body, + "timeout": timeout, + "follow_redirects": False, + } else: - request_kwargs = {"params": body or None, "timeout": timeout} + request_kwargs = { + "params": body or None, + "timeout": timeout, + "follow_redirects": False, + } if headers: request_kwargs["headers"] = headers return request_kwargs @@ -270,7 +278,7 @@ def _next_search_request( method = str(next_link.get("method", "GET")).upper() if method not in {"GET", "POST"}: - return None + raise ValueError(f"STAC pagination link uses an unsupported method: {method!r}") linked_headers = next_link.get("headers") # A server-provided next link may carry continuation credentials. Forward diff --git a/tests/test_stac_server_async.py b/tests/test_stac_server_async.py index 4d291b8..2088eb6 100644 --- a/tests/test_stac_server_async.py +++ b/tests/test_stac_server_async.py @@ -129,6 +129,65 @@ async def handler(request: httpx.Request) -> httpx.Response: assert requests == ["https://stac.example/search"] +@pytest.mark.asyncio +async def test_async_resolver_does_not_follow_redirects() -> None: + requests: list[str] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + requests.append(str(request.url)) + return httpx.Response( + 302, + headers={"Location": "https://attacker.example/collect"}, + request=request, + ) + + async with httpx.AsyncClient( + transport=httpx.MockTransport(handler), follow_redirects=True + ) as client: + with pytest.raises(httpx.HTTPStatusError): + await stac_server.aresolve_cid_from_stac_server( + collection=COLLECTION, + dataset=DATASET, + server_url="https://stac.example", + client=client, + ) + + assert requests == ["https://stac.example/search"] + + +@pytest.mark.asyncio +async def test_async_resolver_rejects_unsupported_pagination_method() -> None: + requests: list[str] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + requests.append(str(request.url)) + return httpx.Response( + 200, + json={ + "features": [_feature("other_dataset", "default", "bafy-other")], + "links": [ + { + "rel": "next", + "href": "/search?page=2", + "method": "DELETE", + } + ], + }, + request=request, + ) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + with pytest.raises(ValueError, match="unsupported method: 'DELETE'"): + await stac_server.aresolve_cid_from_stac_server( + collection=COLLECTION, + dataset=DATASET, + server_url="https://stac.example", + client=client, + ) + + assert requests == ["https://stac.example/search"] + + @pytest.mark.asyncio async def test_async_resolver_drops_linked_headers_on_plaintext_pagination() -> None: requests: list[httpx.Request] = [] From 9504ac047e365e9b757d9c3e0e1678aad4586748 Mon Sep 17 00:00:00 2001 From: 0xSwego <0xSwego@gmail.com> Date: Thu, 6 Aug 2026 15:10:33 +0100 Subject: [PATCH 4/5] docs: keep client libraries in parity --- AGENTS.md | 11 +++++++++++ 1 file changed, 11 insertions(+) create mode 100644 AGENTS.md diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..2ac695f --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,11 @@ +# AGENTS.md + +## Cross-client parity + +This Python client and [dClimate/dclimate-client-js](https://github.com/dClimate/dclimate-client-js) are sibling libraries. Keep their user-visible capabilities and behavior aligned unless a language or runtime difference makes a change inapplicable. + +- For every public API or behavior change—especially STAC/IPFS resolution, dataset loading and selection, metadata, errors, and catalog listing—inspect the corresponding implementation, tests, documentation, and relevant open work in the JavaScript client before finishing. +- Unless the user explicitly limits the task to one repository, treat an applicable sibling-library update as part of the same task. Add equivalent tests and documentation in both projects, using idiomatic APIs for each language rather than mechanically copying implementation details. +- If a change is not applicable to the sibling, or the sibling cannot be updated in the current task, state the reason and leave a concrete follow-up in the handoff or pull-request description. Do not silently allow accidental divergence. +- When reviewing either client, treat undocumented behavioral differences as possible defects and verify whether parity should be restored. + From 2dd7168b890f16d9fe72a2b4b58565ab3dc3e040 Mon Sep 17 00:00:00 2001 From: 0xSwego <0xSwego@gmail.com> Date: Thu, 6 Aug 2026 15:13:59 +0100 Subject: [PATCH 5/5] fix: normalize agents file ending --- AGENTS.md | 1 - 1 file changed, 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index 2ac695f..f1b72f6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -8,4 +8,3 @@ This Python client and [dClimate/dclimate-client-js](https://github.com/dClimate - Unless the user explicitly limits the task to one repository, treat an applicable sibling-library update as part of the same task. Add equivalent tests and documentation in both projects, using idiomatic APIs for each language rather than mechanically copying implementation details. - If a change is not applicable to the sibling, or the sibling cannot be updated in the current task, state the reason and leave a concrete follow-up in the handoff or pull-request description. Do not silently allow accidental divergence. - When reviewing either client, treat undocumented behavioral differences as possible defects and verify whether parity should be restored. -