feat: streaming per-period Icechunk ingest — full #64 implementation

AGENTS.md

@@ -8,7 +8,7 @@ The DHIS2 Climate API is a FastAPI-based REST API that downloads, processes, and
 
 Key concepts:
 
-- **Dataset templates** — YAML files in `data/datasets/` describing a data source (variable, period type, download function). These are blueprints.
+- **Dataset templates** — YAML files in `data/datasets/` describing a data source (variable, period type, ingestion plugin). These are blueprints.
 - **Artifacts / managed datasets** — ingested instances of a template for a specific spatial extent and time range. Exposed under `/datasets` and `/zarr/{dataset_id}`.
 - **Extent** — a single named spatial bounding box configured at instance setup time (`id`, `bbox`, optional `country_code`). Exposed at `GET /extent`.
 - **GeoZarr stores** — datasets are stored as chunked Zarr v3 archives with GeoZarr spatial attributes. Flat stores for small extents; multiscale pyramids for large ones. Served chunk-by-chunk over HTTP with no specialised server middleware.
@@ -44,18 +44,17 @@ The `.env` file is required for `make run` and `make openapi`. Copy `.env.exampl
 
 ## Dataset templates
 
-Each YAML in `data/datasets/` defines a dataset template. The `ingestion` block controls download and zarr build behaviour:
+Each YAML in `data/datasets/` defines a dataset template. The `ingestion` block specifies the plugin class that streams data directly into the Icechunk store:
 
 ```yaml
 ingestion:
-  function: dhis2eo.data.worldpop.pop_total.yearly.download
-  default_params: {} # passed to the download function
+  plugin: climate_api.ingest.plugins.worldpop.WorldPopPlugin
+  params:
+    version: global2
 ```
 
 `build_dataset_zarr` in `data_manager/downloader.py` builds a multiscale Zarr pyramid when the spatial dimensions exceed 2048×2048 pixels; otherwise it writes a flat chunked zarr with chunk sizes derived from the dataset's temporal resolution.
 
-The ingestion interface is being redesigned as a plugin protocol (see GitHub issue #64) — the `ingestion.function` convention will be replaced by a three-method async plugin (`probe`, `periods`, `fetch_period`).
-
 ## pygeoapi
 
 pygeoapi is mounted at `/ogcapi` as a sub-application. Its config is generated dynamically from published artifacts by `publications/services.py` and written to `data/pygeoapi/pygeoapi-config.yml`.

climate_api/client.py

@@ -1,19 +1,13 @@
 """Lightweight client for discovering and opening published Climate API datasets."""
 
-import os
 from urllib.parse import urlparse
 
 import httpx
 import xarray as xr
 
-_FALLBACK_BASE_URL = "http://127.0.0.1:8000"
 _DEFAULT_TIMEOUT = 30.0
 
 
-def _default_base_url() -> str:
-    return os.environ.get("CLIMATE_API_BASE_URL", _FALLBACK_BASE_URL)
-
-
 def _id_from_href(href: str) -> str:
     """Extract the dataset id from a STAC child href by reading the last URL path segment."""
     return urlparse(href).path.rstrip("/").rsplit("/", 1)[-1]
@@ -71,8 +65,8 @@ def open(self, dataset_id: str) -> xr.Dataset:
         """Open a published dataset as an xarray Dataset.
 
         Fetches the STAC collection for ``dataset_id``, reads the Zarr asset
-        metadata, and returns the opened dataset. Coordinates are always
-        ``time``, ``latitude``, and ``longitude``.
+        metadata, and returns the opened dataset. Spatial dimensions are always
+        named ``x`` and ``y``; the time dimension, when present, is ``time``.
         """
         response = self._http.get(f"{self.base_url}/stac/collections/{dataset_id}")
         response.raise_for_status()
@@ -92,55 +86,3 @@ def open(self, dataset_id: str) -> xr.Dataset:
         if not isinstance(open_kwargs, dict):
             raise ValueError(f"Zarr asset for '{dataset_id}' has a malformed xarray:open_kwargs field")
         return xr.open_zarr(href, **open_kwargs)  # type: ignore[no-any-return]
-
-
-def list_datasets(base_url: str | None = None) -> list[dict]:
-    """Return all published datasets from the STAC catalog.
-
-    Each entry is a STAC child link dict with at least ``id``, ``title``, and ``href``.
-    ``base_url`` defaults to the ``CLIMATE_API_BASE_URL`` environment variable,
-    falling back to ``http://127.0.0.1:8000``.
-    """
-    url = (base_url or _default_base_url()).rstrip("/")
-    response = httpx.get(f"{url}/stac/catalog.json", timeout=_DEFAULT_TIMEOUT)
-    response.raise_for_status()
-    catalog = response.json()
-    raw_links = catalog.get("links")
-    if not isinstance(raw_links, list):
-        raise ValueError(f"Invalid STAC catalog response from {url}: missing or non-list 'links' field")
-    links = []
-    for link in raw_links:
-        if isinstance(link, dict) and link.get("rel") == "child":
-            href = link.get("href")
-            if not isinstance(href, str) or not href:
-                raise ValueError(f"STAC child link from {url} has a missing or invalid href")
-            links.append({**link, "id": _id_from_href(href)})
-    return links
-
-
-def open_dataset(dataset_id: str, *, base_url: str | None = None) -> xr.Dataset:
-    """Open a published dataset as an xarray Dataset.
-
-    Fetches the STAC collection for ``dataset_id``, reads the Zarr asset
-    metadata, and returns the opened dataset. Coordinates are always
-    ``time``, ``latitude``, and ``longitude``.
-    ``base_url`` defaults to the ``CLIMATE_API_BASE_URL`` environment variable,
-    falling back to ``http://127.0.0.1:8000``.
-    """
-    url = (base_url or _default_base_url()).rstrip("/")
-    response = httpx.get(f"{url}/stac/collections/{dataset_id}", timeout=_DEFAULT_TIMEOUT)
-    response.raise_for_status()
-    collection = response.json()
-    assets = collection.get("assets")
-    if not isinstance(assets, dict):
-        raise ValueError(f"STAC collection for '{dataset_id}' from {url} has a missing or invalid 'assets' field")
-    asset = assets.get("zarr")
-    if not isinstance(asset, dict):
-        raise ValueError(f"Dataset '{dataset_id}' has no Zarr asset in the STAC collection")
-    href = asset.get("href")
-    if not isinstance(href, str) or not href:
-        raise ValueError(f"Zarr asset for '{dataset_id}' has a missing or invalid href")
-    open_kwargs = asset.get("xarray:open_kwargs", {})
-    if not isinstance(open_kwargs, dict):
-        raise ValueError(f"Zarr asset for '{dataset_id}' has a malformed xarray:open_kwargs field")
-    return xr.open_zarr(href, **open_kwargs)  # type: ignore[no-any-return]

climate_api/data/datasets/chirps3.yaml

@@ -6,16 +6,17 @@
   sync:
     kind: temporal
     execution: append
-    availability:
-      latest_available_function: climate_api.providers.availability.chirps3_daily_latest_available
   extents:
     spatial:
       bbox: [-180, -50, 180, 50]
     temporal:
       begin: "1981-01-01"
       resolution: P1D
   ingestion:
-    function: dhis2eo.data.chc.chirps3.daily.download
+    plugin: climate_api.ingest.plugins.chirps3.Chirps3Plugin
+    params:
+      stage: final
+      flavor: rnl
   units: mm
   resolution: 5 km x 5 km
   source: CHIRPS v3

climate_api/data/datasets/era5_land.yaml

@@ -6,19 +6,16 @@
   sync:
     kind: temporal
     execution: append
-    availability:
-      latest_available_function: climate_api.providers.availability.lagged_latest_available
-      lag_hours: 120
   extents:
     spatial:
       bbox: [-180, -90, 180, 90]
     temporal:
       begin: "1950-01-01"
       resolution: PT1H
   ingestion:
-    function: dhis2eo.data.destine.era5_land.hourly.download
-    default_params:
-      variables: ['t2m']
+    plugin: climate_api.ingest.plugins.era5_land.Era5LandPlugin
+    params:
+      variable: t2m
   transforms:
     - climate_api.transforms.kelvin_to_celsius
   units: degC
@@ -37,19 +34,16 @@
   sync:
     kind: temporal
     execution: append
-    availability:
-      latest_available_function: climate_api.providers.availability.lagged_latest_available
-      lag_hours: 120
   extents:
     spatial:
       bbox: [-180, -90, 180, 90]
     temporal:
       begin: "1950-01-01"
       resolution: PT1H
   ingestion:
-    function: dhis2eo.data.destine.era5_land.hourly.download
-    default_params:
-      variables: ['tp']
+    plugin: climate_api.ingest.plugins.era5_land.Era5LandPlugin
+    params:
+      variable: tp
   transforms:
     - climate_api.transforms.metres_to_mm
   units: mm

climate_api/data/datasets/worldpop.yaml

@@ -5,10 +5,6 @@
   period_type: yearly
   sync:
     kind: release
-    availability:
-      latest_available_function: climate_api.providers.availability.worldpop_release_latest_available
-      # WorldPop projections are intentionally request-driven for future years.
-      allow_future: true
   extents:
     spatial:
       bbox: [-180, -90, 180, 90]
@@ -17,8 +13,9 @@
       end: "2030"
       resolution: P1Y
   ingestion:
-    function: dhis2eo.data.worldpop.pop_total.yearly.download
-    default_params:
+    plugin: climate_api.ingest.plugins.worldpop.WorldPopPlugin
+    params:
+      # country_code is injected automatically from extent.country_code in climate-api.yaml
       version: global2
   units: people
   resolution: 100m x 100m
@@ -27,4 +24,3 @@
   display:
     colormap: reds
     range: [0.0, 25.0]
-    nodata: 0.0

climate_api/data_accessor/services/accessor.py

@@ -3,13 +3,14 @@
 import logging
 import os
 import tempfile
+from pathlib import Path
 from typing import Any
 
 import numpy as np
 import xarray as xr
 from pyproj import Transformer
 
-from ...data_manager.services.downloader import get_cache_files, get_zarr_path
+from ...data_manager.services.downloader import get_icechunk_path
 from ...data_manager.services.utils import get_time_dim, get_x_y_dims
 from ...shared.time import numpy_datetime_to_period_string
 
@@ -24,21 +25,8 @@ def get_data(
 ) -> xr.Dataset:
     """Load an xarray raster dataset for a given time range and bbox."""
     logger.info("Opening dataset")
-    zarr_path = get_zarr_path(dataset)
-    if zarr_path:
-        logger.info(f"Using optimized zarr file: {zarr_path}")
-        ds = open_zarr_dataset(str(zarr_path))
-    else:
-        logger.warning(
-            f"Could not find optimized zarr file for dataset {dataset['id']}, using slower netcdf files instead."
-        )
-        files = get_cache_files(dataset)
-        ds = xr.open_mfdataset(
-            files,
-            data_vars="minimal",
-            coords="minimal",  # pyright: ignore[reportArgumentType]
-            compat="override",
-        )
+    store_path = get_icechunk_path(dataset)
+    ds = open_icechunk_dataset(store_path)
 
     if start and end:
         logger.info(f"Subsetting time to {start} and {end}")
@@ -72,25 +60,10 @@ def get_data_coverage(dataset: dict[str, Any]) -> dict[str, Any]:
 def get_data_coverage_for_paths(
     dataset: dict[str, Any],
     *,
-    zarr_path: str | None = None,
-    netcdf_paths: list[str] | None = None,
+    zarr_path: str,
 ) -> dict[str, Any]:
-    """Return coverage metadata for the concrete files created for one artifact."""
-    if zarr_path is not None and netcdf_paths:
-        raise ValueError("Provide either zarr_path or netcdf_paths when computing coverage, not both")
-    if zarr_path is None and not netcdf_paths:
-        raise ValueError("Coverage calculation requires either zarr_path or at least one netcdf path")
-
-    if zarr_path is not None:
-        ds = open_zarr_dataset(zarr_path)
-    else:
-        assert netcdf_paths is not None
-        ds = xr.open_mfdataset(
-            netcdf_paths,
-            data_vars="minimal",
-            coords="minimal",  # pyright: ignore[reportArgumentType]
-            compat="override",
-        )
+    """Return coverage metadata for a materialized flat-zarr artifact."""
+    ds = open_zarr_dataset(zarr_path)
 
     from climate_api import config as api_config
 
@@ -124,11 +97,40 @@ def open_zarr_dataset(zarr_path: str) -> xr.Dataset:
     return ds
 
 
+def open_icechunk_dataset(store_path: str | Path) -> xr.Dataset:
+    """Open an Icechunk store as an xarray Dataset via a readonly MVCC session.
+
+    Detects multiscale pyramid stores (root group has ``multiscales`` in attrs)
+    and opens group ``0`` (full resolution) in that case.
+    """
+    import icechunk
+    import zarr
+
+    path = Path(store_path)
+    if not path.exists():
+        raise FileNotFoundError(f"Icechunk store not found: {path}")
+    storage = icechunk.local_filesystem_storage(str(path))
+    repo = icechunk.Repository.open(storage)
+    session = repo.readonly_session("main")
+    root = zarr.open_group(session.store, mode="r")
+    group: str | None = "0" if "multiscales" in root.attrs else None
+    return xr.open_zarr(session.store, group=group)  # type: ignore[no-any-return]
+
+
 def _open_zarr(zarr_path: str) -> xr.Dataset:
     """Open a zarr store with automatic consolidated metadata detection."""
     return xr.open_zarr(zarr_path, consolidated=None)  # type: ignore[no-any-return]
 
 
+def coverage_from_open_dataset(ds: xr.Dataset, *, period_type: str, native_crs: str = "EPSG:4326") -> dict[str, Any]:
+    """Summarize temporal and spatial coverage for a caller-managed open dataset.
+
+    Unlike get_data_coverage_for_paths, this function does not close the dataset.
+    Use when the caller already holds a store handle (e.g. an Icechunk session store).
+    """
+    return _coverage_from_dataset(ds=ds, period_type=period_type, native_crs=native_crs)
+
+
 def _coverage_from_dataset(*, ds: xr.Dataset, period_type: str, native_crs: str = "EPSG:4326") -> dict[str, Any]:
     """Summarize temporal and spatial coverage for an already opened dataset."""
     if any(size == 0 for size in ds.sizes.values()):
@@ -141,11 +143,16 @@ def _coverage_from_dataset(*, ds: xr.Dataset, period_type: str, native_crs: str
             },
         }
 
-    time_dim = get_time_dim(ds)
     x_dim, y_dim = get_x_y_dims(ds)
 
-    start = _period_string_scalar(numpy_datetime_to_period_string(ds[time_dim].min(), period_type))  # type: ignore[arg-type]
-    end = _period_string_scalar(numpy_datetime_to_period_string(ds[time_dim].max(), period_type))  # type: ignore[arg-type]
+    start: str | None
+    end: str | None
+    try:
+        time_dim = get_time_dim(ds)
+        start = _period_string_scalar(numpy_datetime_to_period_string(ds[time_dim].min(), period_type))  # type: ignore[arg-type]
+        end = _period_string_scalar(numpy_datetime_to_period_string(ds[time_dim].max(), period_type))  # type: ignore[arg-type]
+    except ValueError:
+        start = end = None
 
     xmin, xmax = ds[x_dim].min().item(), ds[x_dim].max().item()
     ymin, ymax = ds[y_dim].min().item(), ds[y_dim].max().item()

climate_api/data_manager/__init__.py

@@ -1,4 +1,3 @@
 """Data manager package."""
 
-from . import routes as routes
 from . import services as services