diff --git a/README.md b/README.md index 3460d4b..6df49eb 100644 --- a/README.md +++ b/README.md @@ -162,6 +162,71 @@ async def resolve_cid(): ``` +### Dataset version history + +For datasets that advertise version history in STAC, the client follows the +item's `dclimate:versions_api` URL. This automatically selects Hydrogen, +Tritium, or another future version service without a client-side routing map. + +```python +async def list_aigfs_versions(): + client = dClimateClient() + versions = await client.list_dataset_versions( + collection="noaa_aigfs", + dataset="wind_u_forecast", + variant="operational", + anchored=True, + ) + for release in versions.versions: + print(release.version_label, release.cid) + + exact_version = await client.get_dataset_version( + collection="noaa_aigfs", + dataset="wind_u_forecast", + variant="operational", + commit_id="commit-id", + ) + print(exact_version.cid) +``` + +The lower-level functions in `dclimate_client_py.ceramic_api` retain their +existing names and explicit `base_url` support. STAC-aware applications should +prefer `list_dataset_versions()` and `get_dataset_version()` so they do not need +to know which service owns a dataset. + +### Multiresolution datasets + +Pyramidal datasets require an explicit resolution (recommended) or raw Zarr +group. The client reports the available resolutions instead of silently +choosing between different precision, chunking, and fetching strategies. + +```python +data, metadata = await client.load_dataset( + collection="copernicus_clms", + dataset="fpar", + resolution="2km", +) +print(metadata["resolution"], metadata["zarr_group"]) +``` + +FPAR's advertised mappings are `500m` → group `"0"`, `2km` → group `"1"`, +and `8km` → group `"2"`. For example, replace `resolution="2km"` above with +`"500m"` or `"8km"` to select those levels. Raw `zarr_group="1"` is supported +when a caller intentionally works at the storage level, but do not pass it +together with `resolution`. + +STAC may temporarily include a legacy `assets.data` alias for 500 m alongside +the three named assets. The client ignores that alias when enumerating choices, +so it is not a default or a fourth resolution. Consumers that previously +relied on `assets.data` or implicit group `"0"` should migrate to an explicit +resolution before the alias is removed in a future breaking release. + +Callers loading a direct CID have no STAC resolution mapping and must pass +`zarr_group` when the store contains multiple groups; human-readable +`resolution` is rejected because the mapping exists only in STAC. The STAC +generator's `metadataGroup` is internal catalog-generation configuration and +does not influence client selection. + ## Siren API usage The Python client also exposes a Siren REST client for metrics and regions. diff --git a/dclimate_client_py/__init__.py b/dclimate_client_py/__init__.py index 20eb826..0586b56 100644 --- a/dclimate_client_py/__init__.py +++ b/dclimate_client_py/__init__.py @@ -14,6 +14,7 @@ ) from .stac_server import ( ResolvedDataset, + ZarrResolution, aclose_stac_server_client, aresolve_cid_from_stac_server, resolve_cid_from_stac_server, @@ -33,6 +34,9 @@ EvmSigner, ) from .dclimate_zarr_errors import ( + ConflictingResolutionSelectionError, + MultiresolutionSelectionRequiredError, + ResolutionNotAvailableError, SirenApiError, X402PaymentError, X402NotInstalledError, @@ -78,11 +82,15 @@ def __dir__() -> list[str]: "load_stac_catalog", "list_available_datasets", "ResolvedDataset", + "ZarrResolution", "aclose_stac_server_client", "aresolve_cid_from_stac_server", "resolve_cid_from_stac_server", "list_available_datasets_from_stac_server", "STAC_SERVER_URL", + "MultiresolutionSelectionRequiredError", + "ResolutionNotAvailableError", + "ConflictingResolutionSelectionError", # Siren "SirenClient", "SirenApiKeyAuth", diff --git a/dclimate_client_py/ceramic_api.py b/dclimate_client_py/ceramic_api.py new file mode 100644 index 0000000..b0bfac6 --- /dev/null +++ b/dclimate_client_py/ceramic_api.py @@ -0,0 +1,314 @@ +""" +Ceramic dataset version and provenance helpers. + +This module provides a small Python interface for the dClimate Ceramic API so +consumers can list dataset versions, select exact historical releases, resolve +citations, and build gateway URLs for reproducible access workflows. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional +from urllib.parse import quote, urlsplit, urlunsplit + +import httpx + + +HYDROGEN_CERAMIC_API_BASE_URL = "https://hydrogen.dclimate.net/api" +TRITIUM_CERAMIC_API_BASE_URL = "https://tritium.dclimate.net/api" +# Kept for callers that explicitly rely on the legacy default. STAC-aware code +# should follow the complete dclimate:versions_api URL instead. +DEFAULT_CERAMIC_API_BASE_URL = HYDROGEN_CERAMIC_API_BASE_URL +DEFAULT_IPFS_GATEWAY_BASE_URL = "https://ipfs-gateway.dclimate.net" + + +@dataclass(frozen=True) +class VerificationInfo: + """Verification metadata derived from Ceramic anchoring state.""" + + anchor_status: Optional[str] = None + details: Dict[str, Any] = field(default_factory=dict) + + @classmethod + def from_api_payload(cls, payload: Optional[Dict[str, Any]]) -> "VerificationInfo": + payload = payload or {} + details = dict(payload) + anchor_status = details.pop("anchorStatus", None) + return cls(anchor_status=anchor_status, details=details) + + +@dataclass(frozen=True) +class DatasetVersion: + """One versioned dataset snapshot returned by the Ceramic API.""" + + dataset: str + cid: str + old_cid: Optional[str] = None + timestamp: Optional[int] = None + stream_id: Optional[str] = None + commit_id: Optional[str] = None + controller_did: Optional[str] = None + published_at: Optional[str] = None + version_label: Optional[str] = None + release_class: Optional[str] = None + is_citable: Optional[bool] = None + retention_class: Optional[str] = None + verification: VerificationInfo = field(default_factory=VerificationInfo) + + @classmethod + def from_api_payload(cls, payload: Dict[str, Any]) -> "DatasetVersion": + return cls( + dataset=payload["dataset"], + cid=payload["cid"], + old_cid=payload.get("oldCid"), + timestamp=payload.get("timestamp"), + stream_id=payload.get("streamId"), + commit_id=payload.get("commitId"), + controller_did=payload.get("controllerDid"), + published_at=payload.get("publishedAt"), + version_label=payload.get("versionLabel"), + release_class=payload.get("releaseClass"), + is_citable=payload.get("isCitable"), + retention_class=payload.get("retentionClass"), + verification=VerificationInfo.from_api_payload(payload.get("verification")), + ) + + +@dataclass(frozen=True) +class DatasetVersionListing: + """Version-history response for one dataset.""" + + dataset: str + stream_id: Optional[str] + versions: List[DatasetVersion] + + +@dataclass(frozen=True) +class CitationInfo: + """Citation payload for one dataset release.""" + + dataset: str + stream_id: Optional[str] + commit_id: Optional[str] + cid: str + published_at: Optional[str] + version_label: Optional[str] + is_citable: Optional[bool] + retention_class: Optional[str] + citation: str + + @classmethod + def from_api_payload(cls, payload: Dict[str, Any]) -> "CitationInfo": + return cls( + dataset=payload["dataset"], + stream_id=payload.get("streamId"), + commit_id=payload.get("commitId"), + cid=payload["cid"], + published_at=payload.get("publishedAt"), + version_label=payload.get("versionLabel"), + is_citable=payload.get("isCitable"), + retention_class=payload.get("retentionClass"), + citation=payload["citation"], + ) + + +def _normalize_base_url(base_url: str) -> str: + return base_url.rstrip("/") + + +def _dataset_path(dataset: str) -> str: + return quote(dataset, safe="") + + +def _request_json( + url: str, + params: Optional[Dict[str, Any]] = None, + session: Optional[httpx.Client] = None, +) -> Dict[str, Any]: + if session is not None: + response = session.get(url, params=params, timeout=30) + response.raise_for_status() + return response.json() + with httpx.Client(timeout=30, follow_redirects=True) as client: + response = client.get(url, params=params) + response.raise_for_status() + return response.json() + + +def _encode_bool(value: Optional[bool]) -> Optional[str]: + if value is None: + return None + return str(value).lower() + + +def _append_url_path(url: str, component: str) -> str: + """Append one encoded path component without disturbing URL query data.""" + parts = urlsplit(url) + path = f"{parts.path.rstrip('/')}/{quote(component, safe='')}" + return urlunsplit((parts.scheme, parts.netloc, path, parts.query, parts.fragment)) + + +def list_versions_from_url( + versions_url: str, + anchored: Optional[bool] = None, + is_citable: Optional[bool] = None, + version_label: Optional[str] = None, + session: Optional[httpx.Client] = None, +) -> DatasetVersionListing: + """List versions using the complete service URL discovered from STAC.""" + params: Dict[str, Any] = {} + encoded_anchored = _encode_bool(anchored) + encoded_is_citable = _encode_bool(is_citable) + if encoded_anchored is not None: + params["anchored"] = encoded_anchored + if encoded_is_citable is not None: + params["isCitable"] = encoded_is_citable + if version_label is not None: + params["versionLabel"] = version_label + + payload = _request_json( + versions_url, + params=params or None, + session=session, + ) + versions = [ + DatasetVersion.from_api_payload(item) for item in payload.get("versions", []) + ] + return DatasetVersionListing( + dataset=payload["dataset"], + stream_id=payload.get("streamId"), + versions=versions, + ) + + +def get_exact_version_from_url( + versions_url: str, + commit_id: str, + session: Optional[httpx.Client] = None, +) -> DatasetVersion: + """Resolve an exact release from a STAC-discovered versions URL.""" + payload = _request_json( + _append_url_path(versions_url, commit_id), + session=session, + ) + return DatasetVersion.from_api_payload(payload) + + +def get_citation_from_url( + citation_url: str, + session: Optional[httpx.Client] = None, +) -> CitationInfo: + """Fetch citation metadata using the complete URL discovered from STAC.""" + return CitationInfo.from_api_payload(_request_json(citation_url, session=session)) + + +def list_versions( + dataset: str, + base_url: str = DEFAULT_CERAMIC_API_BASE_URL, + anchored: Optional[bool] = None, + is_citable: Optional[bool] = None, + version_label: Optional[str] = None, + session: Optional[httpx.Client] = None, +) -> DatasetVersionListing: + """ + List known versions for one dataset. + """ + dataset_path = _dataset_path(dataset) + return list_versions_from_url( + f"{_normalize_base_url(base_url)}/datasets/{dataset_path}/versions", + anchored=anchored, + is_citable=is_citable, + version_label=version_label, + session=session, + ) + + +def get_exact_version( + dataset: str, + commit_id: str, + base_url: str = DEFAULT_CERAMIC_API_BASE_URL, + session: Optional[httpx.Client] = None, +) -> DatasetVersion: + """ + Resolve one exact dataset version by commit id. + """ + dataset_path = _dataset_path(dataset) + commit_path = quote(commit_id, safe="") + payload = _request_json( + f"{_normalize_base_url(base_url)}/datasets/{dataset_path}/versions/{commit_path}", + session=session, + ) + return DatasetVersion.from_api_payload(payload) + + +def get_latest_metadata( + dataset: str, + base_url: str = DEFAULT_CERAMIC_API_BASE_URL, + session: Optional[httpx.Client] = None, +) -> DatasetVersion: + """ + Get the latest dataset metadata from the Ceramic API. + """ + dataset_path = _dataset_path(dataset) + payload = _request_json( + f"{_normalize_base_url(base_url)}/datasets/{dataset_path}", + session=session, + ) + return DatasetVersion.from_api_payload(payload) + + +def get_citation( + dataset: str, + commit_id: Optional[str] = None, + base_url: str = DEFAULT_CERAMIC_API_BASE_URL, + session: Optional[httpx.Client] = None, +) -> CitationInfo: + """ + Fetch citation metadata for a dataset release. + """ + params = {"commitId": commit_id} if commit_id else None + dataset_path = _dataset_path(dataset) + payload = _request_json( + f"{_normalize_base_url(base_url)}/datasets/{dataset_path}/citation", + params=params, + session=session, + ) + return CitationInfo.from_api_payload(payload) + + +def filter_anchored_versions(versions: List[DatasetVersion]) -> List[DatasetVersion]: + """Return only anchored versions from a version list.""" + return [ + version + for version in versions + if version.verification.anchor_status == "anchored" + ] + + +def get_latest_anchored_version( + dataset: str, + base_url: str = DEFAULT_CERAMIC_API_BASE_URL, + session: Optional[httpx.Client] = None, +) -> DatasetVersion: + """ + Return the most recent anchored version for one dataset. + """ + listing = list_versions( + dataset=dataset, + base_url=base_url, + anchored=True, + session=session, + ) + anchored_versions = filter_anchored_versions(listing.versions) + if not anchored_versions: + raise ValueError(f"No anchored versions found for dataset '{dataset}'") + return anchored_versions[-1] + + +def build_gateway_url( + cid: str, + gateway_base: str = DEFAULT_IPFS_GATEWAY_BASE_URL, +) -> str: + """Build a gateway URL for one IPFS CID.""" + return f"{gateway_base.rstrip('/')}/ipfs/{cid}" diff --git a/dclimate_client_py/datasets.py b/dclimate_client_py/datasets.py index c7d41bf..21d7f26 100644 --- a/dclimate_client_py/datasets.py +++ b/dclimate_client_py/datasets.py @@ -15,8 +15,6 @@ # --- Type Definitions --- -hydrogen_endpoint = "https://dclimate-ceramic.duckdns.org/api/datasets" - class SpatialExtent(TypedDict): """Bounding box for a dataset's spatial coverage.""" @@ -97,6 +95,15 @@ class DatasetMetadata(TypedDict, total=False): ] # How the dataset was loaded organization: Optional[str] zarr_group: Optional[str] + resolution: Optional[str] + versions_api: Optional[str] + provenance_api: Optional[str] + citation_api: Optional[str] + stream_id: Optional[str] + commit_id: Optional[str] + version_label: Optional[str] + is_citable: Optional[bool] + retention_class: Optional[str] # --- Helper Functions --- diff --git a/dclimate_client_py/dclimate_client.py b/dclimate_client_py/dclimate_client.py index 230cf6c..2e55e1a 100644 --- a/dclimate_client_py/dclimate_client.py +++ b/dclimate_client_py/dclimate_client.py @@ -22,12 +22,23 @@ from .geotemporal_data import GeotemporalData from .datasets import DatasetMetadata -from .dclimate_zarr_errors import InvalidSelectionError +from .dclimate_zarr_errors import ( + ConflictingResolutionSelectionError, + InvalidSelectionError, + MultiresolutionSelectionRequiredError, + ResolutionNotAvailableError, +) from .stac_server import ( - ResolvedDataset, - aresolve_cid_from_stac_server, + ResolvedDatasetDetails, + aresolve_dataset_from_stac_server, list_available_datasets_from_stac_server, ) +from .ceramic_api import ( + DatasetVersion, + DatasetVersionListing, + get_exact_version_from_url, + list_versions_from_url, +) from .siren import SirenClient from .siren.types import ( SirenMetricDataPoint, @@ -240,6 +251,62 @@ def _apply_zarr_group_metadata(ds: xr.Dataset, metadata: DatasetMetadata) -> Non if isinstance(loaded_zarr_group, str): metadata["zarr_group"] = loaded_zarr_group + @staticmethod + def _resolve_zarr_selection( + resolved: ResolvedDatasetDetails, + *, + resolution: typing.Optional[str], + zarr_group: typing.Optional[str], + ) -> tuple[typing.Optional[str], typing.Optional[str]]: + if resolution is not None and zarr_group is not None: + raise ConflictingResolutionSelectionError( + "Pass either resolution or zarr_group, not both." + ) + + choices = resolved.zarr_resolutions + if resolution is not None: + match = next( + (choice for choice in choices if choice.resolution == resolution), + None, + ) + if match is None: + available = tuple(choice.resolution for choice in choices) + raise ResolutionNotAvailableError( + f"Resolution '{resolution}' is not available." + + (f" Choose one of: {', '.join(available)}." if available else "") + ) + return match.group, match.resolution + + if zarr_group is not None: + normalized_group = zarr_group.strip("/") + if choices and normalized_group not in {choice.group for choice in choices}: + available_groups = tuple(choice.group for choice in choices) + raise ResolutionNotAvailableError( + f"Zarr group '{normalized_group}' is not available. " + f"Choose one of: {', '.join(available_groups)}." + ) + selected_resolution = next( + ( + choice.resolution + for choice in choices + if choice.group == normalized_group + ), + None, + ) + return normalized_group, selected_resolution + + if len(choices) > 1: + available = tuple(choice.resolution for choice in choices) + raise MultiresolutionSelectionRequiredError( + "This dataset has multiple resolutions; pass resolution or zarr_group. " + f"Available resolutions: {', '.join(available)}.", + available_resolutions=available, + available_groups=tuple(choice.group for choice in choices), + ) + if len(choices) == 1: + return choices[0].group, choices[0].resolution + return None, None + async def load_dataset( self, dataset: str, @@ -250,6 +317,7 @@ async def load_dataset( return_xarray: bool = False, zarr_group: typing.Optional[str] = None, shard_read_mode: typing.Literal["full", "sparse"] = "sparse", + resolution: typing.Optional[str] = None, ) -> typing.Union[ typing.Tuple[GeotemporalData, DatasetMetadata], typing.Tuple[xr.Dataset, DatasetMetadata], @@ -282,9 +350,10 @@ async def load_dataset( If True, return raw xarray.Dataset. If False (default), return GeotemporalData wrapper. zarr_group : str, optional - Explicit Zarr group to open for grouped/pyramid sharded stores. If - omitted, py-hamt v2 stores with multiple top-level groups default - to group "0" when available. + Explicit Zarr group to open for grouped/pyramid sharded stores. + resolution : str, optional + Human-readable resolution advertised by STAC. Multiresolution + datasets require either this parameter or ``zarr_group``. shard_read_mode : {"full", "sparse"}, optional Sharded Zarr shard-index read strategy. Defaults to ``"sparse"`` and decodes only the requested shard slot on read-only cache @@ -336,6 +405,14 @@ async def load_dataset( # Case 1: Direct CID provided - bypass catalog resolution metadata: DatasetMetadata if cid: + if resolution is not None and zarr_group is not None: + raise ConflictingResolutionSelectionError( + "Pass either resolution or zarr_group, not both." + ) + if resolution is not None: + raise ResolutionNotAvailableError( + "resolution requires STAC metadata; pass zarr_group for a direct CID." + ) direct_collection = collection if ( organization @@ -391,12 +468,12 @@ async def load_dataset( if organization and not collection.startswith(f"{organization}_"): resolved_collection = f"{organization}_{collection}" - resolved: typing.Optional[ResolvedDataset] = None + resolved: typing.Optional[ResolvedDatasetDetails] = None # Try STAC server first (faster, avoids loading IPFS catalog) if self._stac_server_url: try: - resolved = await aresolve_cid_from_stac_server( + resolved = await aresolve_dataset_from_stac_server( collection=resolved_collection, dataset=dataset, variant=variant, @@ -412,7 +489,7 @@ async def load_dataset( from .stac_catalog import ( list_available_datasets, load_stac_catalog, - resolve_dataset_cid_from_stac, + resolve_dataset_from_stac, ) # Lazy load STAC catalog @@ -440,7 +517,7 @@ async def load_dataset( resolved_collection = prefixed_matches[0] resolved = await asyncio.to_thread( - resolve_dataset_cid_from_stac, + resolve_dataset_from_stac, catalog=self._stac_catalog, collection=resolved_collection, dataset=dataset, @@ -450,10 +527,16 @@ async def load_dataset( assert resolved is not None + selected_group, selected_resolution = self._resolve_zarr_selection( + resolved, + resolution=resolution, + zarr_group=zarr_group, + ) + ds = await _load_dataset_from_ipfs_cid( ipfs_cid=resolved.cid, kubo_cas=self._kubo_cas, - zarr_group=zarr_group, + zarr_group=selected_group, shard_read_mode=shard_read_mode, ) @@ -478,7 +561,25 @@ async def load_dataset( else None ), } + if resolved.versions_api is not None: + metadata["versions_api"] = resolved.versions_api + if resolved.provenance_api is not None: + metadata["provenance_api"] = resolved.provenance_api + if resolved.citation_api is not None: + metadata["citation_api"] = resolved.citation_api + if resolved.stream_id is not None: + metadata["stream_id"] = resolved.stream_id + if resolved.commit_id is not None: + metadata["commit_id"] = resolved.commit_id + if resolved.version_label is not None: + metadata["version_label"] = resolved.version_label + if resolved.is_citable is not None: + metadata["is_citable"] = resolved.is_citable + if resolved.retention_class is not None: + metadata["retention_class"] = resolved.retention_class self._apply_zarr_group_metadata(ds, metadata) + if selected_resolution is not None: + metadata["resolution"] = selected_resolution if return_xarray: return ds, metadata @@ -617,6 +718,136 @@ async def alist_datasets(self) -> typing.Dict[str, typing.Dict[str, typing.Any]] return await asyncio.to_thread(list_available_datasets, self._stac_catalog) + async def _resolve_dataset_details( + self, + collection: str, + dataset: str, + variant: typing.Optional[str], + organization: typing.Optional[str], + ) -> ResolvedDatasetDetails: + """Resolve release metadata through the hosted STAC API or IPFS fallback.""" + resolved_collection = collection + if organization and not collection.startswith(f"{organization}_"): + resolved_collection = f"{organization}_{collection}" + + if self._stac_server_url: + try: + if self._kubo_cas is None: + async with httpx.AsyncClient( + timeout=30, follow_redirects=False + ) as client: + return await aresolve_dataset_from_stac_server( + collection=resolved_collection, + dataset=dataset, + variant=variant, + server_url=self._stac_server_url, + client=client, + ) + return await aresolve_dataset_from_stac_server( + collection=resolved_collection, + dataset=dataset, + variant=variant, + server_url=self._stac_server_url, + client=self._get_stac_http_client(), + ) + except (httpx.HTTPError, ValueError): + pass + + from .stac_catalog import ( + list_available_datasets, + load_stac_catalog, + resolve_dataset_from_stac, + ) + + if self._stac_catalog is None: + async with self._stac_catalog_lock: + if self._stac_catalog is None: + self._stac_catalog = await asyncio.to_thread( + load_stac_catalog, + gateway_url=self._catalog_gateway_base_url, + headers=self._headers, + auth=self._auth, + ) + + if not organization and resolved_collection: + available = await asyncio.to_thread( + list_available_datasets, self._stac_catalog + ) + if resolved_collection not in available: + prefixed_matches = [ + coll_id + for coll_id in available + if coll_id.endswith(f"_{resolved_collection}") + ] + if len(prefixed_matches) == 1: + resolved_collection = prefixed_matches[0] + + return await asyncio.to_thread( + resolve_dataset_from_stac, + catalog=self._stac_catalog, + collection=resolved_collection, + dataset=dataset, + variant=variant, + organization=organization, + ) + + async def list_dataset_versions( + self, + collection: str, + dataset: str, + variant: typing.Optional[str] = None, + organization: typing.Optional[str] = None, + *, + anchored: typing.Optional[bool] = None, + is_citable: typing.Optional[bool] = None, + version_label: typing.Optional[str] = None, + ) -> DatasetVersionListing: + """List releases using the version-service URL advertised by STAC.""" + details = await self._resolve_dataset_details( + collection=collection, + dataset=dataset, + variant=variant, + organization=organization, + ) + if not details.versions_api: + raise ValueError( + "Version history is not available for " + f"{collection}/{dataset}/{details.variant}" + ) + return await asyncio.to_thread( + list_versions_from_url, + details.versions_api, + anchored=anchored, + is_citable=is_citable, + version_label=version_label, + ) + + async def get_dataset_version( + self, + collection: str, + dataset: str, + commit_id: str, + variant: typing.Optional[str] = None, + organization: typing.Optional[str] = None, + ) -> DatasetVersion: + """Resolve one exact release through the version URL advertised by STAC.""" + details = await self._resolve_dataset_details( + collection=collection, + dataset=dataset, + variant=variant, + organization=organization, + ) + if not details.versions_api: + raise ValueError( + "Version history is not available for " + f"{collection}/{dataset}/{details.variant}" + ) + return await asyncio.to_thread( + get_exact_version_from_url, + details.versions_api, + commit_id, + ) + # ------------------------------------------------------------------ # Siren REST API methods # ------------------------------------------------------------------ diff --git a/dclimate_client_py/dclimate_zarr_errors.py b/dclimate_client_py/dclimate_zarr_errors.py index 7a4abb7..c23f5a8 100644 --- a/dclimate_client_py/dclimate_zarr_errors.py +++ b/dclimate_client_py/dclimate_zarr_errors.py @@ -66,6 +66,29 @@ class InvalidSelectionError(ZarrClientError): """Raised when dataset/collection/variant selection is invalid or ambiguous""" +class MultiresolutionSelectionRequiredError(InvalidSelectionError): + """Raised when a pyramidal dataset requires an explicit resolution or group.""" + + def __init__( + self, + message: str, + *, + available_resolutions: tuple[str, ...] = (), + available_groups: tuple[str, ...] = (), + ) -> None: + super().__init__(message) + self.available_resolutions = available_resolutions + self.available_groups = available_groups + + +class ResolutionNotAvailableError(InvalidSelectionError): + """Raised when a requested resolution is not advertised by STAC.""" + + +class ConflictingResolutionSelectionError(InvalidSelectionError): + """Raised when both resolution and raw Zarr group are provided.""" + + class VariantNotFoundError(ZarrClientError): """Raised when specified variant is not found in dataset""" diff --git a/dclimate_client_py/ipfs_retrieval.py b/dclimate_client_py/ipfs_retrieval.py index 45fe5a8..681e8c4 100644 --- a/dclimate_client_py/ipfs_retrieval.py +++ b/dclimate_client_py/ipfs_retrieval.py @@ -40,6 +40,7 @@ def observe(name: str, seconds: float) -> None: from .dclimate_zarr_errors import ( IpfsConnectionError, + MultiresolutionSelectionRequiredError, ) # Configure logging @@ -174,23 +175,20 @@ def _normalize_zarr_group(zarr_group: str | None) -> str | None: return normalized or None -def _zarr_group_candidates(store: Any) -> list[str]: - """Return safe default Zarr groups from py-hamt v2 stores.""" +def _available_zarr_groups(store: Any) -> tuple[str, ...]: + """Return the Zarr groups reported by a py-hamt v2 store.""" groups_getter = getattr(store, "_v2_top_level_groups", None) if not callable(groups_getter): - return [] + return () try: groups = groups_getter() except Exception: - return [] + return () - normalized_groups = sorted( - group.strip("/") for group in groups if isinstance(group, str) and group + return tuple( + sorted(group.strip("/") for group in groups if isinstance(group, str) and group) ) - if "0" not in normalized_groups: - return [] - return ["0"] def _store_requires_explicit_zarr_group(store: Any) -> bool: @@ -216,7 +214,7 @@ def _open_zarr_from_store( *, zarr_group: str | None = None, ) -> tuple[xr.Dataset, str | None]: - """Open a Zarr store, choosing a default group for py-hamt v2 pyramids.""" + """Open a Zarr store without silently selecting a pyramid resolution.""" normalized_group = _normalize_zarr_group(zarr_group) if normalized_group is not None: return ( @@ -225,23 +223,24 @@ def _open_zarr_from_store( ) if _store_requires_explicit_zarr_group(store): - for candidate_group in _zarr_group_candidates(store): - return ( - xr.open_zarr(store=store, group=candidate_group, decode_timedelta=True), - candidate_group, - ) + groups = _available_zarr_groups(store) + raise MultiresolutionSelectionRequiredError( + "This Zarr store has multiple groups; pass zarr_group explicitly." + + (f" Available groups: {', '.join(groups)}." if groups else ""), + available_groups=groups, + ) try: return xr.open_zarr(store=store, decode_timedelta=True), None except ValueError as exc: if not _is_explicit_zarr_group_error(exc): raise - for candidate_group in _zarr_group_candidates(store): - return ( - xr.open_zarr(store=store, group=candidate_group, decode_timedelta=True), - candidate_group, - ) - raise + groups = _available_zarr_groups(store) + raise MultiresolutionSelectionRequiredError( + "This Zarr store requires an explicit zarr_group." + + (f" Available groups: {', '.join(groups)}." if groups else ""), + available_groups=groups, + ) from exc async def _open_sharded_zarr_store( @@ -457,8 +456,8 @@ async def _load_dataset_from_ipfs_cid( dataset_status = "connection_error" _record_span_error(dataset_span, exc) raise - except ValueError as exc: - # Re-raise ValueError as-is + except (ValueError, MultiresolutionSelectionRequiredError) as exc: + # Preserve selection and Zarr value errors as-is. _record_span_error(dataset_span, exc) raise except Exception as e: diff --git a/dclimate_client_py/stac_catalog.py b/dclimate_client_py/stac_catalog.py index 1146387..0691b6a 100644 --- a/dclimate_client_py/stac_catalog.py +++ b/dclimate_client_py/stac_catalog.py @@ -18,6 +18,9 @@ from .datasets import SpatialExtent, TemporalExtent from .stac_server import ( ResolvedDataset, + ResolvedDatasetDetails, + ZarrResolution, + _dedupe_zarr_resolutions, _dataset_and_variant_from_item_id, _dataset_and_variant_from_known_datasets, ) @@ -290,17 +293,17 @@ def load_stac_catalog( return catalog -def resolve_dataset_cid_from_stac( +def resolve_dataset_from_stac( catalog: pystac.Catalog, collection: str, dataset: str, variant: Optional[str] = None, organization: Optional[str] = None, -) -> ResolvedDataset: +) -> ResolvedDatasetDetails: """ Resolve a dataset to its IPFS CID by querying the STAC catalog. - Changed in 0.6: returns ResolvedDataset. + Changed in 0.6: returns ResolvedDatasetDetails. This function navigates the STAC catalog structure to find the specific dataset variant and extracts the Zarr data CID from the STAC Item's assets. @@ -318,7 +321,7 @@ def resolve_dataset_cid_from_stac( catalog metadata. Returns: - ResolvedDataset: The IPFS CID and selected variant + ResolvedDatasetDetails: CID, selected variant, and release-service metadata Raises: ValueError: If collection, dataset, or variant is not found in the catalog @@ -457,13 +460,61 @@ def resolve_dataset_cid_from_stac( break assert selected_variant is not None - if "data" in selected_item.assets: - href = selected_item.assets["data"].href + properties = selected_item.properties or {} + zarr_resolutions = _dedupe_zarr_resolutions( + ZarrResolution( + asset_key=asset_key, + resolution=asset.extra_fields["dclimate:spatial_resolution"], + group=asset.extra_fields["dclimate:zarr_group"], + ) + for asset_key, asset in selected_item.assets.items() + if asset_key != "data" + and isinstance(asset.extra_fields.get("dclimate:spatial_resolution"), str) + and isinstance(asset.extra_fields.get("dclimate:zarr_group"), str) + ) + data_asset = selected_item.assets.get("data") + selected_asset = data_asset or ( + selected_item.assets[zarr_resolutions[0].asset_key] + if zarr_resolutions + else None + ) + if selected_asset is not None: + href = selected_asset.href if href.startswith("ipfs://"): href = href.replace("ipfs://", "") - return ResolvedDataset(href, selected_variant) + return ResolvedDatasetDetails( + cid=href, + variant=selected_variant, + versions_api=properties.get("dclimate:versions_api"), + provenance_api=properties.get("dclimate:provenance_api"), + citation_api=properties.get("dclimate:citation_api"), + stream_id=properties.get("dclimate:stream_id"), + commit_id=properties.get("dclimate:commit_id"), + version_label=properties.get("dclimate:version_label"), + is_citable=properties.get("dclimate:is_citable"), + retention_class=properties.get("dclimate:retention_class"), + zarr_resolutions=zarr_resolutions, + ) + + raise ValueError(f"Item '{selected_item.id}' does not have a readable data asset") + - raise ValueError(f"Item '{selected_item.id}' does not have a 'data' asset") +def resolve_dataset_cid_from_stac( + catalog: pystac.Catalog, + collection: str, + dataset: str, + variant: Optional[str] = None, + organization: Optional[str] = None, +) -> ResolvedDataset: + """Resolve a CID while preserving the original two-field public result.""" + details = resolve_dataset_from_stac( + catalog=catalog, + collection=collection, + dataset=dataset, + variant=variant, + organization=organization, + ) + return details.as_resolved_dataset() def _extract_item_extents( diff --git a/dclimate_client_py/stac_server.py b/dclimate_client_py/stac_server.py index 8eadb2d..b09d9a4 100644 --- a/dclimate_client_py/stac_server.py +++ b/dclimate_client_py/stac_server.py @@ -8,6 +8,7 @@ import asyncio import weakref from collections.abc import AsyncIterator, Iterator +from dataclasses import dataclass from json import dumps from threading import Lock from typing import Any, Dict, Iterable, NamedTuple, Optional, Set @@ -70,6 +71,49 @@ class ResolvedDataset(NamedTuple): variant: str +@dataclass(frozen=True) +class ZarrResolution: + """One explicitly selectable resolution asset advertised by STAC.""" + + asset_key: str + resolution: str + group: str + + +def _dedupe_zarr_resolutions( + resolutions: Iterable[ZarrResolution], +) -> tuple[ZarrResolution, ...]: + """Deduplicate equivalent resolution/group mappings, preserving asset order.""" + seen: set[tuple[str, str]] = set() + unique: list[ZarrResolution] = [] + for choice in resolutions: + key = (choice.resolution, choice.group) + if key not in seen: + seen.add(key) + unique.append(choice) + return tuple(unique) + + +@dataclass(frozen=True) +class ResolvedDatasetDetails: + """Dataset location and optional release services advertised by STAC.""" + + cid: str + variant: str + versions_api: Optional[str] = None + provenance_api: Optional[str] = None + citation_api: Optional[str] = None + stream_id: Optional[str] = None + commit_id: Optional[str] = None + version_label: Optional[str] = None + is_citable: Optional[bool] = None + retention_class: Optional[str] = None + zarr_resolutions: tuple[ZarrResolution, ...] = () + + def as_resolved_dataset(self) -> ResolvedDataset: + return ResolvedDataset(self.cid, self.variant) + + def _dataset_and_variant_from_item_id( feature_id: str, collection: str, @@ -403,7 +447,7 @@ def _resolve_dataset_from_features( dataset: str, variant: Optional[str], features: Iterable[Dict[str, Any]], -) -> ResolvedDataset: +) -> ResolvedDatasetDetails: """Resolve a dataset from STAC search features shared by both clients.""" feature_list = list(features) @@ -456,21 +500,51 @@ def _effective_variant(feature: Dict[str, Any]) -> str: # Extract CID from asset selected_variant = variant if variant is not None else _effective_variant(item) - href = item.get("assets", {}).get("data", {}).get("href", "") - if href.startswith("ipfs://"): - return ResolvedDataset(href.replace("ipfs://", ""), selected_variant) - if href: - return ResolvedDataset(href, selected_variant) + properties = item.get("properties") or {} + assets = item.get("assets", {}) + data_asset = assets.get("data", {}) + zarr_resolutions = _dedupe_zarr_resolutions( + ZarrResolution( + asset_key=asset_key, + resolution=asset["dclimate:spatial_resolution"], + group=asset["dclimate:zarr_group"], + ) + for asset_key, asset in assets.items() + if asset_key != "data" + and isinstance(asset, dict) + and isinstance(asset.get("dclimate:spatial_resolution"), str) + and isinstance(asset.get("dclimate:zarr_group"), str) + ) + href = data_asset.get("href", "") + cid = _strip_ipfs_scheme(href) or _strip_ipfs_scheme( + properties.get("dclimate:latest_dataset_cid") + ) + if not cid and zarr_resolutions: + cid = _strip_ipfs_scheme(assets[zarr_resolutions[0].asset_key].get("href")) + if cid: + return ResolvedDatasetDetails( + cid=cid, + variant=selected_variant, + versions_api=properties.get("dclimate:versions_api"), + provenance_api=properties.get("dclimate:provenance_api"), + citation_api=properties.get("dclimate:citation_api"), + stream_id=properties.get("dclimate:stream_id"), + commit_id=properties.get("dclimate:commit_id"), + version_label=properties.get("dclimate:version_label"), + is_citable=properties.get("dclimate:is_citable"), + retention_class=properties.get("dclimate:retention_class"), + zarr_resolutions=zarr_resolutions, + ) raise ValueError(f"Item '{item['id']}' has no data asset") -def resolve_cid_from_stac_server( +def resolve_dataset_from_stac_server( collection: str, dataset: str, variant: Optional[str] = None, server_url: str = STAC_SERVER_URL, -) -> ResolvedDataset: +) -> ResolvedDatasetDetails: """ Resolve dataset CID via STAC server /search API. @@ -486,7 +560,7 @@ def resolve_cid_from_stac_server( server_url: STAC server base URL Returns: - ResolvedDataset: The IPFS CID and selected variant + ResolvedDatasetDetails: CID, selected variant, and release-service metadata Raises: ValueError: If dataset or variant is not found @@ -504,20 +578,20 @@ def resolve_cid_from_stac_server( return _resolve_dataset_from_features(collection, dataset, variant, features) -async def aresolve_cid_from_stac_server( +async def aresolve_dataset_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. +) -> ResolvedDatasetDetails: + """Resolve dataset details natively asynchronously via the STAC API. - When ``client`` is omitted, calls reuse a pooled ``httpx.AsyncClient`` - 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. + Async counterpart of ``resolve_dataset_from_stac_server``. When ``client`` + is omitted, calls reuse a pooled ``httpx.AsyncClient`` 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') @@ -527,7 +601,7 @@ async def aresolve_cid_from_stac_server( client: Optional caller-owned pooled async HTTP client Returns: - ResolvedDataset: The IPFS CID and selected variant + ResolvedDatasetDetails: CID, selected variant, and release-service metadata Raises: ValueError: If dataset or variant is not found @@ -543,6 +617,43 @@ async def aresolve_cid_from_stac_server( return _resolve_dataset_from_features(collection, dataset, variant, features) +def resolve_cid_from_stac_server( + collection: str, + dataset: str, + variant: Optional[str] = None, + server_url: str = STAC_SERVER_URL, +) -> ResolvedDataset: + """Resolve a CID while preserving the original two-field public result.""" + return resolve_dataset_from_stac_server( + collection=collection, + dataset=dataset, + variant=variant, + server_url=server_url, + ).as_resolved_dataset() + + +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: + """Async counterpart of ``resolve_cid_from_stac_server``. + + See ``aresolve_dataset_from_stac_server`` for client-ownership semantics. + """ + resolved = await aresolve_dataset_from_stac_server( + collection=collection, + dataset=dataset, + variant=variant, + server_url=server_url, + client=client, + ) + return resolved.as_resolved_dataset() + + def _strip_ipfs_scheme(cid: Optional[str]) -> Optional[str]: if not cid: return None diff --git a/pyproject.toml b/pyproject.toml index 609b65f..53385ce 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,7 @@ build-backend = "pdm.backend" [project] name = "dclimate-client-py" -version = "0.7.0" # Set a static version or handle it in versioning strategy +version = "0.8.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/test_ceramic_api.py b/tests/test_ceramic_api.py new file mode 100644 index 0000000..973888f --- /dev/null +++ b/tests/test_ceramic_api.py @@ -0,0 +1,294 @@ +import unittest +from unittest.mock import Mock + +from dclimate_client_py.ceramic_api import ( + CitationInfo, + DatasetVersion, + DatasetVersionListing, + build_gateway_url, + filter_anchored_versions, + get_citation, + get_citation_from_url, + get_exact_version, + get_exact_version_from_url, + get_latest_anchored_version, + get_latest_metadata, + list_versions, + list_versions_from_url, +) + + +def _mock_response(payload): + response = Mock() + response.json.return_value = payload + response.raise_for_status.return_value = None + return response + + +class CeramicApiTests(unittest.TestCase): + def test_list_versions_builds_filters_and_parses_response(self): + session = Mock() + session.get.return_value = _mock_response( + { + "dataset": "eagle-temp", + "streamId": "stream-123", + "versions": [ + { + "dataset": "eagle-temp", + "cid": "cid-1", + "commitId": "commit-1", + "streamId": "stream-123", + "versionLabel": "2026-05-01", + "verification": {"anchorStatus": "anchored"}, + }, + { + "dataset": "eagle-temp", + "cid": "cid-2", + "commitId": "commit-2", + "streamId": "stream-123", + "versionLabel": "2026-05-02", + "verification": {"anchorStatus": "pending"}, + }, + ], + } + ) + + result = list_versions( + "eagle-temp", + anchored=True, + is_citable=False, + version_label="2026-05", + session=session, + ) + + self.assertIsInstance(result, DatasetVersionListing) + self.assertEqual(result.dataset, "eagle-temp") + self.assertEqual(result.stream_id, "stream-123") + self.assertEqual(len(result.versions), 2) + self.assertEqual(result.versions[0].verification.anchor_status, "anchored") + session.get.assert_called_once_with( + "https://hydrogen.dclimate.net/api/datasets/eagle-temp/versions", + params={ + "anchored": "true", + "isCitable": "false", + "versionLabel": "2026-05", + }, + timeout=30, + ) + + def test_get_exact_version_returns_dataset_version(self): + session = Mock() + session.get.return_value = _mock_response( + { + "dataset": "eagle-temp", + "cid": "cid-1", + "commitId": "commit-1", + "streamId": "stream-123", + "publishedAt": "2026-05-02T14:26:43.131Z", + "versionLabel": "2026-05", + "isCitable": True, + "retentionClass": "permanent", + "verification": {"anchorStatus": "anchored"}, + } + ) + + result = get_exact_version("eagle-temp", "commit-1", session=session) + + self.assertIsInstance(result, DatasetVersion) + self.assertEqual(result.dataset, "eagle-temp") + self.assertEqual(result.commit_id, "commit-1") + self.assertEqual(result.verification.anchor_status, "anchored") + session.get.assert_called_once_with( + "https://hydrogen.dclimate.net/api/datasets/eagle-temp/versions/commit-1", + params=None, + timeout=30, + ) + + def test_get_latest_metadata_uses_latest_dataset_endpoint(self): + session = Mock() + session.get.return_value = _mock_response( + { + "dataset": "eagle-temp", + "cid": "cid-latest", + "commitId": "commit-latest", + "streamId": "stream-123", + "verification": {"anchorStatus": "anchored"}, + } + ) + + result = get_latest_metadata("eagle-temp", session=session) + + self.assertEqual(result.cid, "cid-latest") + session.get.assert_called_once_with( + "https://hydrogen.dclimate.net/api/datasets/eagle-temp", + params=None, + timeout=30, + ) + + def test_get_citation_returns_citation_info(self): + session = Mock() + session.get.return_value = _mock_response( + { + "dataset": "eagle-temp", + "streamId": "stream-123", + "commitId": "commit-1", + "cid": "cid-1", + "publishedAt": "2026-05-02T14:26:43.131Z", + "versionLabel": "2026-05", + "isCitable": True, + "retentionClass": "permanent", + "citation": "citation text", + } + ) + + result = get_citation("eagle-temp", commit_id="commit-1", session=session) + + self.assertIsInstance(result, CitationInfo) + self.assertEqual(result.citation, "citation text") + session.get.assert_called_once_with( + "https://hydrogen.dclimate.net/api/datasets/eagle-temp/citation", + params={"commitId": "commit-1"}, + timeout=30, + ) + + def test_filter_anchored_versions_keeps_only_anchored_entries(self): + # Build versions through parsing so the verification model matches runtime code. + parsed_versions = [ + DatasetVersion.from_api_payload( + { + "dataset": "eagle-temp", + "cid": "cid-1", + "commitId": "commit-1", + "verification": {"anchorStatus": "anchored"}, + } + ), + DatasetVersion.from_api_payload( + { + "dataset": "eagle-temp", + "cid": "cid-2", + "commitId": "commit-2", + "verification": {"anchorStatus": "pending"}, + } + ), + ] + + anchored = filter_anchored_versions(parsed_versions) + + self.assertEqual(len(anchored), 1) + self.assertEqual(anchored[0].commit_id, "commit-1") + + def test_get_latest_anchored_version_preserves_service_event_order(self): + session = Mock() + session.get.return_value = _mock_response( + { + "dataset": "eagle-temp", + "streamId": "stream-123", + "versions": [ + { + "dataset": "eagle-temp", + "cid": "cid-1", + "commitId": "commit-1", + "timestamp": 200, + "verification": {"anchorStatus": "anchored"}, + }, + { + "dataset": "eagle-temp", + "cid": "cid-2", + "commitId": "commit-2", + "timestamp": 100, + "verification": {"anchorStatus": "anchored"}, + }, + ], + } + ) + + result = get_latest_anchored_version("eagle-temp", session=session) + + self.assertEqual(result.commit_id, "commit-2") + session.get.assert_called_once_with( + "https://hydrogen.dclimate.net/api/datasets/eagle-temp/versions", + params={"anchored": "true"}, + timeout=30, + ) + + def test_get_latest_anchored_version_raises_when_no_anchored_versions_exist(self): + session = Mock() + session.get.return_value = _mock_response( + { + "dataset": "eagle-temp", + "streamId": "stream-123", + "versions": [], + } + ) + + with self.assertRaisesRegex( + ValueError, "No anchored versions found for dataset 'eagle-temp'" + ): + get_latest_anchored_version("eagle-temp", session=session) + + def test_build_gateway_url_joins_gateway_and_cid(self): + self.assertEqual( + build_gateway_url("bafytest", "https://gateway.example.com/"), + "https://gateway.example.com/ipfs/bafytest", + ) + + def test_list_versions_from_stac_url_preserves_tritium_dataset_slug(self): + session = Mock() + session.get.return_value = _mock_response( + {"dataset": "era5-temperature-2m-finalized", "versions": []} + ) + + result = list_versions_from_url( + "https://tritium.dclimate.net/api/datasets/era5-temperature-2m-finalized/versions", + anchored=True, + session=session, + ) + + self.assertEqual(result.dataset, "era5-temperature-2m-finalized") + session.get.assert_called_once_with( + "https://tritium.dclimate.net/api/datasets/era5-temperature-2m-finalized/versions", + params={"anchored": "true"}, + timeout=30, + ) + + def test_get_exact_version_from_stac_url_encodes_commit(self): + session = Mock() + session.get.return_value = _mock_response( + {"dataset": "aigfs-wind-u", "cid": "cid-1"} + ) + + result = get_exact_version_from_url( + "https://hydrogen.dclimate.net/api/datasets/aigfs-wind-u/versions", + "commit/one", + session=session, + ) + + self.assertEqual(result.cid, "cid-1") + session.get.assert_called_once_with( + "https://hydrogen.dclimate.net/api/datasets/aigfs-wind-u/versions/commit%2Fone", + params=None, + timeout=30, + ) + + def test_get_citation_from_stac_url_preserves_commit_query(self): + session = Mock() + citation_url = ( + "https://hydrogen.dclimate.net/api/datasets/aigfs-wind-u/citation" + "?commitId=commit-1" + ) + session.get.return_value = _mock_response( + { + "dataset": "aigfs-wind-u", + "cid": "cid-1", + "citation": "citation text", + } + ) + + result = get_citation_from_url(citation_url, session=session) + + self.assertEqual(result.citation, "citation text") + session.get.assert_called_once_with(citation_url, params=None, timeout=30) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_ipfs_retrieval.py b/tests/test_ipfs_retrieval.py index 8b59c3c..93d9780 100644 --- a/tests/test_ipfs_retrieval.py +++ b/tests/test_ipfs_retrieval.py @@ -98,32 +98,23 @@ async def hamt_build(**kwargs): @pytest.mark.asyncio -async def test_multigroup_sharded_store_defaults_to_group_zero(monkeypatch): +async def test_multigroup_sharded_store_requires_explicit_group(monkeypatch): open_kwargs = [] async def sharded_open(**kwargs): open_kwargs.append(kwargs) return DummyGroupedStore() - opened_groups = [] - - def open_zarr(*, store, group=None, decode_timedelta=False): - assert decode_timedelta is True - opened_groups.append(group) - return xr.Dataset() - monkeypatch.setattr(ipfs_retrieval.ShardedZarrStore, "open", sharded_open) - monkeypatch.setattr(ipfs_retrieval.xr, "open_zarr", open_zarr) - ds = await ipfs_retrieval._load_dataset_from_ipfs_cid( - VALID_CID, - DummyKuboCAS(), - ) + with pytest.raises(ipfs_retrieval.MultiresolutionSelectionRequiredError) as raised: + await ipfs_retrieval._load_dataset_from_ipfs_cid( + VALID_CID, + DummyKuboCAS(), + ) - assert opened_groups == ["0"] + assert raised.value.available_groups == ("0", "1") assert open_kwargs[0]["shard_read_mode"] == "sparse" - assert ds.attrs["_ipfs_store_type"] == "ShardedZarrStore" - assert ds.attrs["_ipfs_zarr_group"] == "0" @pytest.mark.asyncio @@ -167,7 +158,10 @@ def open_zarr(*, store, group=None, decode_timedelta=False): monkeypatch.setattr(ipfs_retrieval.HAMT, "build", hamt_build) monkeypatch.setattr(ipfs_retrieval.xr, "open_zarr", open_zarr) - with pytest.raises(ValueError, match="explicit Zarr group"): + with pytest.raises( + ipfs_retrieval.MultiresolutionSelectionRequiredError, + match="explicit zarr_group", + ): await ipfs_retrieval._load_dataset_from_ipfs_cid( VALID_CID, DummyKuboCAS(), @@ -271,3 +265,27 @@ async def load_dataset_from_ipfs_cid( assert isinstance(ds, xr.Dataset) assert metadata["zarr_group"] == "2" + + +@pytest.mark.asyncio +async def test_client_preserves_positional_zarr_group_and_shard_mode(monkeypatch): + observed = [] + + async def load_dataset_from_ipfs_cid(**kwargs): + observed.append((kwargs["zarr_group"], kwargs["shard_read_mode"])) + return xr.Dataset(attrs={"_ipfs_zarr_group": kwargs["zarr_group"]}) + + monkeypatch.setattr( + dclimate_client_module, + "_load_dataset_from_ipfs_cid", + load_dataset_from_ipfs_cid, + ) + dclimate = dClimateClient() + dclimate._kubo_cas = DummyKuboCAS() + + _, metadata = await dclimate.load_dataset( + "pyramid", None, None, None, VALID_CID, True, "2", "full" + ) + + assert observed == [("2", "full")] + assert metadata["zarr_group"] == "2" diff --git a/tests/test_review_fu_httpx.py b/tests/test_review_fu_httpx.py index 5937336..d9f358b 100644 --- a/tests/test_review_fu_httpx.py +++ b/tests/test_review_fu_httpx.py @@ -16,7 +16,7 @@ from dclimate_client_py import stac_catalog from dclimate_client_py.dclimate_client import dClimateClient from dclimate_client_py.stac_catalog import IPFSStacIO -from dclimate_client_py.stac_server import ResolvedDataset +from dclimate_client_py.stac_server import ResolvedDatasetDetails REPO_ROOT = Path(__file__).resolve().parents[1] @@ -105,13 +105,13 @@ def fake_from_file( lambda loaded_catalog: {"review_collection": {"types": ["temperature"]}}, ) - def resolve_from_catalog(**kwargs: Any) -> ResolvedDataset: + def resolve_from_catalog(**kwargs: Any) -> ResolvedDatasetDetails: catalog_resolutions.append(kwargs) - return ResolvedDataset("bafy-fallback-dataset", "default") + return ResolvedDatasetDetails("bafy-fallback-dataset", "default") monkeypatch.setattr( stac_catalog, - "resolve_dataset_cid_from_stac", + "resolve_dataset_from_stac", resolve_from_catalog, ) diff --git a/tests/test_stac_server_async.py b/tests/test_stac_server_async.py index 2088eb6..1eb4bdc 100644 --- a/tests/test_stac_server_async.py +++ b/tests/test_stac_server_async.py @@ -1,13 +1,17 @@ from __future__ import annotations -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, Mock 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 +from dclimate_client_py.stac_server import ( + ResolvedDataset, + ResolvedDatasetDetails, + ZarrResolution, +) COLLECTION = "example_collection" @@ -322,17 +326,127 @@ async def handler(request: httpx.Request) -> httpx.Response: assert resolved == ResolvedDataset("bafy-default", "default") +@pytest.mark.asyncio +async def test_async_details_resolver_preserves_resolution_choices() -> None: + feature = _feature(DATASET, "default", "bafy-pyramid") + feature["assets"] = { + "data": {"href": "ipfs://bafy-pyramid"}, + "data-500m": { + "href": "ipfs://bafy-pyramid", + "dclimate:zarr_group": "0", + "dclimate:spatial_resolution": "500m", + }, + "data-2km": { + "href": "ipfs://bafy-pyramid", + "dclimate:zarr_group": "1", + "dclimate:spatial_resolution": "2km", + }, + } + + async def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"features": [feature]}, request=request) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + resolved = await stac_server.aresolve_dataset_from_stac_server( + collection=COLLECTION, + dataset=DATASET, + server_url="https://stac.example", + client=client, + ) + + assert resolved.cid == "bafy-pyramid" + assert resolved.zarr_resolutions == ( + ZarrResolution("data-500m", "500m", "0"), + ZarrResolution("data-2km", "2km", "1"), + ) + + +@pytest.mark.asyncio +async def test_version_resolution_closes_temporary_client_outside_context( + monkeypatch: pytest.MonkeyPatch, +) -> None: + temporary_client = AsyncMock(spec=httpx.AsyncClient) + temporary_client.__aenter__.return_value = temporary_client + client_factory = Mock(return_value=temporary_client) + resolver = AsyncMock( + return_value=ResolvedDatasetDetails( + cid="bafy-versioned", + variant="default", + versions_api="https://versions.test/datasets/example/versions", + ) + ) + + monkeypatch.setattr(dclimate_client.httpx, "AsyncClient", client_factory) + monkeypatch.setattr(dclimate_client, "aresolve_dataset_from_stac_server", resolver) + client = dclimate_client.dClimateClient(stac_server_url="https://stac.example") + + details = await client._resolve_dataset_details( + collection=COLLECTION, + dataset=DATASET, + variant="default", + organization=None, + ) + + assert details.cid == "bafy-versioned" + resolver.assert_awaited_once_with( + collection=COLLECTION, + dataset=DATASET, + variant="default", + server_url="https://stac.example", + client=temporary_client, + ) + temporary_client.__aenter__.assert_awaited_once() + temporary_client.__aexit__.assert_awaited_once() + assert client._stac_http_client is None + + +@pytest.mark.asyncio +async def test_version_resolution_reuses_context_owned_client( + monkeypatch: pytest.MonkeyPatch, +) -> None: + pooled_client = AsyncMock(spec=httpx.AsyncClient) + pooled_client.is_closed = False + resolver = AsyncMock( + return_value=ResolvedDatasetDetails(cid="bafy-versioned", variant="default") + ) + client_factory = Mock(side_effect=AssertionError("temporary client created")) + + monkeypatch.setattr(dclimate_client.httpx, "AsyncClient", client_factory) + monkeypatch.setattr(dclimate_client, "aresolve_dataset_from_stac_server", resolver) + client = dclimate_client.dClimateClient(stac_server_url="https://stac.example") + client._kubo_cas = object() + client._stac_http_client = pooled_client + + await client._resolve_dataset_details( + collection=COLLECTION, + dataset=DATASET, + variant="default", + organization=None, + ) + + resolver.assert_awaited_once_with( + collection=COLLECTION, + dataset=DATASET, + variant="default", + server_url="https://stac.example", + client=pooled_client, + ) + client_factory.assert_not_called() + + @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")) + resolver = AsyncMock( + return_value=ResolvedDatasetDetails(cid="bafy-native", variant="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, "aresolve_dataset_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() diff --git a/tests/test_stac_version_discovery.py b/tests/test_stac_version_discovery.py new file mode 100644 index 0000000..90d1157 --- /dev/null +++ b/tests/test_stac_version_discovery.py @@ -0,0 +1,421 @@ +from unittest.mock import Mock + +import pytest +import pystac +import xarray as xr +import httpx + +from dclimate_client_py import dclimate_client as client_module +from dclimate_client_py import stac_catalog, stac_server +from dclimate_client_py.ceramic_api import DatasetVersion, DatasetVersionListing +from dclimate_client_py.dclimate_client import dClimateClient +from dclimate_client_py.dclimate_zarr_errors import ( + ConflictingResolutionSelectionError, + MultiresolutionSelectionRequiredError, + ResolutionNotAvailableError, +) + + +def _feature(properties, *, asset_fields=None): + return { + "type": "Feature", + "id": "noaa_aigfs-wind_u_forecast-operational", + "collection": "noaa_aigfs", + "properties": properties, + "assets": {"data": {"href": "ipfs://bafy-current", **(asset_fields or {})}}, + } + + +def test_stac_server_details_preserve_discovered_hydrogen_urls(monkeypatch): + versions_url = "https://hydrogen.dclimate.net/api/datasets/aigfs-wind-u/versions" + feature = _feature( + { + "dclimate:dataset_id": "wind_u_forecast", + "dclimate:variant": "operational", + "dclimate:versions_api": versions_url, + "dclimate:commit_id": "commit-1", + "dclimate:is_citable": False, + "dclimate:retention_class": "ephemeral", + }, + asset_fields={ + "dclimate:zarr_group": "0", + "dclimate:spatial_resolution": "500m", + }, + ) + monkeypatch.setattr( + stac_server, + "_search_pages", + lambda *args, **kwargs: iter([{"features": [feature]}]), + ) + + details = stac_server.resolve_dataset_from_stac_server( + "noaa_aigfs", "wind_u_forecast", "operational" + ) + + assert details.cid == "bafy-current" + assert details.versions_api == versions_url + assert details.commit_id == "commit-1" + assert details.is_citable is False + assert details.retention_class == "ephemeral" + assert details.zarr_resolutions == () + # The old API remains a two-field tuple for backwards compatibility. + assert stac_server.resolve_cid_from_stac_server( + "noaa_aigfs", "wind_u_forecast", "operational" + ) == stac_server.ResolvedDataset("bafy-current", "operational") + + +@pytest.mark.parametrize("include_alias", [True, False]) +def test_stac_server_treats_named_assets_as_three_resolution_choices( + monkeypatch, include_alias +): + assets = { + f"data-{resolution}": { + "href": "ipfs://bafy-fpar", + "dclimate:zarr_group": group, + "dclimate:spatial_resolution": resolution, + } + for resolution, group in (("500m", "0"), ("2km", "1"), ("8km", "2")) + } + assets["data-500m-alias"] = dict(assets["data-500m"]) + if include_alias: + assets["data"] = { + **assets["data-500m"], + "title": "Legacy compatibility alias", + } + feature = _feature( + { + "dclimate:dataset_id": "wind_u_forecast", + "dclimate:variant": "operational", + } + ) + feature["assets"] = assets + monkeypatch.setattr( + stac_server, + "_search_pages", + lambda *args, **kwargs: iter([{"features": [feature]}]), + ) + + details = stac_server.resolve_dataset_from_stac_server( + "noaa_aigfs", "wind_u_forecast", "operational" + ) + + assert len(details.zarr_resolutions) == 3 + assert {choice.asset_key for choice in details.zarr_resolutions} == { + "data-500m", + "data-2km", + "data-8km", + } + + +def test_ipfs_stac_details_preserve_discovered_tritium_url(): + catalog = pystac.Catalog(id="root", description="root") + organization = pystac.Catalog(id="ecmwf", description="ECMWF") + collection = pystac.Collection( + id="ecmwf_era5", + description="ERA5", + extent=pystac.Extent( + pystac.SpatialExtent([[-180, -90, 180, 90]]), + pystac.TemporalExtent([[None, None]]), + ), + ) + item = pystac.Item( + id="ecmwf_era5-temperature_2m-finalized", + geometry=None, + bbox=None, + datetime=None, + properties={ + "start_datetime": "1940-01-01T00:00:00Z", + "end_datetime": "2026-01-01T00:00:00Z", + "dclimate:dataset_id": "temperature_2m", + "dclimate:variant": "finalized", + "dclimate:versions_api": ( + "https://tritium.dclimate.net/api/datasets/" + "era5-temperature-2m-finalized/versions" + ), + }, + ) + data_asset = pystac.Asset(href="ipfs://bafy-era5") + data_asset.extra_fields["dclimate:zarr_group"] = "0" + data_asset.extra_fields["dclimate:spatial_resolution"] = "500m" + item.add_asset("data-500m", data_asset) + duplicate_asset = pystac.Asset(href="ipfs://bafy-era5") + duplicate_asset.extra_fields.update(data_asset.extra_fields) + item.add_asset("data-500m-alias", duplicate_asset) + collection.add_item(item) + collection_link = organization.add_child(collection) + collection_link.extra_fields["dclimate:id"] = "ecmwf_era5" + organization_link = catalog.add_child(organization) + organization_link.extra_fields.update( + { + "dclimate:id": "ecmwf", + "dclimate:collections:historical": ["ecmwf_era5"], + } + ) + + details = stac_catalog.resolve_dataset_from_stac( + catalog, "ecmwf_era5", "temperature_2m", "finalized" + ) + + assert details.cid == "bafy-era5" + assert details.versions_api == ( + "https://tritium.dclimate.net/api/datasets/" + "era5-temperature-2m-finalized/versions" + ) + assert details.zarr_resolutions == ( + stac_server.ZarrResolution("data-500m", "500m", "0"), + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("resolution", "group"), [("500m", None), (None, "/2/")]) +async def test_client_requires_explicit_resolution_or_group( + monkeypatch, resolution, group +): + details = stac_server.ResolvedDatasetDetails( + cid="bafy-grouped", + variant="default", + versions_api="https://versions.test/datasets/pyramid/versions", + provenance_api="https://versions.test/datasets/pyramid/provenance", + citation_api="https://versions.test/datasets/pyramid/citation", + stream_id="stream-1", + commit_id="commit-1", + version_label="2026-08", + is_citable=False, + retention_class="permanent", + zarr_resolutions=( + stac_server.ZarrResolution("data-500m", "500m", "0"), + stac_server.ZarrResolution("data-2km", "2km", "2"), + ), + ) + + async def aresolve(**kwargs): + return details + + monkeypatch.setattr( + client_module, + "aresolve_dataset_from_stac_server", + aresolve, + ) + + observed_groups = [] + + async def load_dataset_from_ipfs_cid(**kwargs): + observed_groups.append(kwargs["zarr_group"]) + normalized = kwargs["zarr_group"] + return xr.Dataset(attrs={"_ipfs_zarr_group": normalized}) + + monkeypatch.setattr( + client_module, + "_load_dataset_from_ipfs_cid", + load_dataset_from_ipfs_cid, + ) + client = dClimateClient(stac_server_url="https://stac.test") + client._kubo_cas = object() + + _, metadata = await client.load_dataset( + dataset="pyramid", + collection="test_grouped", + resolution=resolution, + zarr_group=group, + return_xarray=True, + ) + + expected_group = "0" if resolution == "500m" else "2" + assert observed_groups == [expected_group] + assert metadata["zarr_group"] == expected_group + assert metadata["resolution"] == (resolution or "2km") + assert metadata["versions_api"] == details.versions_api + assert metadata["provenance_api"] == details.provenance_api + assert metadata["citation_api"] == details.citation_api + assert metadata["stream_id"] == "stream-1" + assert metadata["commit_id"] == "commit-1" + assert metadata["version_label"] == "2026-08" + assert metadata["is_citable"] is False + assert metadata["retention_class"] == "permanent" + + +@pytest.mark.asyncio +async def test_client_rejects_ambiguous_or_invalid_resolution_selection(monkeypatch): + details = stac_server.ResolvedDatasetDetails( + cid="bafy-grouped", + variant="default", + zarr_resolutions=( + stac_server.ZarrResolution("data-500m", "500m", "0"), + stac_server.ZarrResolution("data-2km", "2km", "1"), + ), + ) + + async def aresolve(**kwargs): + return details + + monkeypatch.setattr(client_module, "aresolve_dataset_from_stac_server", aresolve) + client = dClimateClient(stac_server_url="https://stac.test") + client._kubo_cas = object() + + with pytest.raises(MultiresolutionSelectionRequiredError) as required: + await client.load_dataset(dataset="pyramid", collection="test_grouped") + assert required.value.available_resolutions == ("500m", "2km") + + with pytest.raises(ResolutionNotAvailableError, match="10km"): + await client.load_dataset( + dataset="pyramid", collection="test_grouped", resolution="10km" + ) + + with pytest.raises(ConflictingResolutionSelectionError): + await client.load_dataset( + dataset="pyramid", + collection="test_grouped", + resolution="500m", + zarr_group="0", + ) + + +@pytest.mark.asyncio +async def test_client_lists_versions_from_stac_url(monkeypatch): + client = dClimateClient() + details = stac_server.ResolvedDatasetDetails( + cid="bafy-current", + variant="operational", + versions_api="https://hydrogen.test/api/datasets/aigfs-wind-u/versions", + ) + + async def resolve_details(*args, **kwargs): + return details + + listing = DatasetVersionListing("aigfs-wind-u", "stream-1", []) + request = Mock(return_value=listing) + monkeypatch.setattr(client, "_resolve_dataset_details", resolve_details) + monkeypatch.setattr( + "dclimate_client_py.dclimate_client.list_versions_from_url", request + ) + + result = await client.list_dataset_versions( + "noaa_aigfs", + "wind_u_forecast", + "operational", + anchored=True, + ) + + assert result is listing + request.assert_called_once_with( + details.versions_api, + anchored=True, + is_citable=None, + version_label=None, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "versions_url", + [ + "https://hydrogen.dclimate.net/api/datasets/aigfs-wind-u/versions", + "https://tritium.dclimate.net/api/datasets/aigfs-wind-u/versions", + ], +) +async def test_client_gets_exact_version_from_stac_url(monkeypatch, versions_url): + client = dClimateClient() + details = stac_server.ResolvedDatasetDetails( + cid="bafy-current", + variant="operational", + versions_api=versions_url, + ) + + async def resolve_details(*args, **kwargs): + return details + + exact = DatasetVersion(dataset="aigfs-wind-u", cid="bafy-exact") + request = Mock(return_value=exact) + monkeypatch.setattr(client, "_resolve_dataset_details", resolve_details) + monkeypatch.setattr(client_module, "get_exact_version_from_url", request) + + result = await client.get_dataset_version( + "noaa_aigfs", + "wind_u_forecast", + "commit/with spaces?and=query#fragment", + "operational", + ) + + assert result is exact + request.assert_called_once_with( + versions_url, "commit/with spaces?and=query#fragment" + ) + + +@pytest.mark.asyncio +async def test_client_propagates_exact_version_http_error(monkeypatch): + client = dClimateClient() + + async def resolve_details(*args, **kwargs): + return stac_server.ResolvedDatasetDetails( + cid="bafy-current", + variant="operational", + versions_api="https://hydrogen.test/datasets/aigfs/versions", + ) + + request = httpx.Request("GET", "https://hydrogen.test/datasets/aigfs/versions/c") + response = httpx.Response(503, request=request) + error = httpx.HTTPStatusError("unavailable", request=request, response=response) + monkeypatch.setattr(client, "_resolve_dataset_details", resolve_details) + monkeypatch.setattr( + client_module, + "get_exact_version_from_url", + Mock(side_effect=error), + ) + + with pytest.raises(httpx.HTTPStatusError) as raised: + await client.get_dataset_version("noaa_aigfs", "wind_u_forecast", "commit-1") + + assert raised.value.response.status_code == 503 + + +@pytest.mark.asyncio +async def test_client_reports_items_without_version_history(monkeypatch): + client = dClimateClient() + + async def resolve_details(*args, **kwargs): + return stac_server.ResolvedDatasetDetails("bafy", "default") + + monkeypatch.setattr(client, "_resolve_dataset_details", resolve_details) + + with pytest.raises(ValueError, match="Version history is not available"): + await client.list_dataset_versions("copernicus_clms", "fpar") + + with pytest.raises(ValueError, match="Version history is not available"): + await client.get_dataset_version("copernicus_clms", "fpar", "commit-1") + + +@pytest.mark.asyncio +async def test_version_catalog_fallback_normalizes_shorthand_collection(monkeypatch): + catalog = object() + resolved = stac_server.ResolvedDatasetDetails( + cid="bafy-era5", + variant="finalized", + versions_api="https://versions.test/era5/versions", + ) + resolver = Mock(return_value=resolved) + + monkeypatch.setattr(stac_catalog, "load_stac_catalog", Mock(return_value=catalog)) + monkeypatch.setattr( + stac_catalog, + "list_available_datasets", + Mock(return_value={"ecmwf_era5": {}}), + ) + monkeypatch.setattr(stac_catalog, "resolve_dataset_from_stac", resolver) + client = dClimateClient(stac_server_url=None) + + details = await client._resolve_dataset_details( + collection="era5", + dataset="temperature_2m", + variant="finalized", + organization=None, + ) + + assert details is resolved + resolver.assert_called_once_with( + catalog=catalog, + collection="ecmwf_era5", + dataset="temperature_2m", + variant="finalized", + organization=None, + )