Catalog & utility API

The shared plumbing every provider backend builds on: catalog loading, the strict YAML parser, the provider registry, and the small filesystem helpers. See Base contracts for the rules these implement.

Catalog loading

All 48 catalog loaders route through load_catalog, which owns the catalog glob, the (path, mtime_ns) cache key, and the cache registry.

earthlens.base.load_catalog(path, cache, parse, *, provider, shard_noun='')

Return the parsed catalog at path, memoised on the files' mtimes.

The composition every provider loader repeats: resolve the contributing files, build the key, return a live cache hit, else call parse and store the result. parse receives the file list and owns everything provider-specific — the row models, the merge across shards, the duplicate-key checks.

Parameters:

Name	Type	Description	Default
path	Path	The catalog directory or single YAML file.	required
cache	CatalogParseCache	The module's :class:CatalogParseCache.	required
parse	Callable[[list[Path]], T]	Callable taking the contributing files and returning the parsed catalog. Called only on a cache miss.	required
provider	str	Provider name for the not-found error.	required
shard_noun	str	Optional sharding description for that error.	''

Returns:

Type	Description
T	Whatever parse returned, from the cache when the mtimes are unchanged.

Raises:

Type	Description
ValueError	If path does not exist (see :func:yaml_files_for).

Examples:

• The parse runs once, then the cached value is reused:

  >>> import tempfile
  >>> from pathlib import Path
  >>> from earthlens.base.yaml_loader import CatalogParseCache
  >>> from earthlens.base.catalog_source import load_catalog
  >>> one = Path(tempfile.mkdtemp()) / "c.yaml"
  >>> _ = one.write_text("datasets: {}\n")
  >>> cache, calls = CatalogParseCache(), []
  >>> def parse(files):
  ...     calls.append(files)
  ...     return {"rows": len(files)}
  >>> load_catalog(one, cache, parse, provider="Demo")
  {'rows': 1}
  >>> load_catalog(one, cache, parse, provider="Demo")
  {'rows': 1}
  >>> len(calls)
  1
  

Source code in libs/core/src/earthlens/base/catalog_source.py

def load_catalog(
    path: Path,
    cache: CatalogParseCache,
    parse: Callable[[list[Path]], T],
    *,
    provider: str,
    shard_noun: str = "",
) -> T:
    """Return the parsed catalog at `path`, memoised on the files' mtimes.

    The composition every provider loader repeats: resolve the contributing
    files, build the key, return a live cache hit, else call `parse` and store
    the result. `parse` receives the file list and owns everything
    provider-specific — the row models, the merge across shards, the
    duplicate-key checks.

    Args:
        path: The catalog directory or single YAML file.
        cache: The module's :class:`CatalogParseCache`.
        parse: Callable taking the contributing files and returning the parsed
            catalog. Called only on a cache miss.
        provider: Provider name for the not-found error.
        shard_noun: Optional sharding description for that error.

    Returns:
        Whatever `parse` returned, from the cache when the mtimes are unchanged.

    Raises:
        ValueError: If `path` does not exist (see :func:`yaml_files_for`).

    Examples:
        - The parse runs once, then the cached value is reused:
            ```python
            >>> import tempfile
            >>> from pathlib import Path
            >>> from earthlens.base.yaml_loader import CatalogParseCache
            >>> from earthlens.base.catalog_source import load_catalog
            >>> one = Path(tempfile.mkdtemp()) / "c.yaml"
            >>> _ = one.write_text("datasets: {}\\n")
            >>> cache, calls = CatalogParseCache(), []
            >>> def parse(files):
            ...     calls.append(files)
            ...     return {"rows": len(files)}
            >>> load_catalog(one, cache, parse, provider="Demo")
            {'rows': 1}
            >>> load_catalog(one, cache, parse, provider="Demo")
            {'rows': 1}
            >>> len(calls)
            1

            ```
    """
    files = yaml_files_for(path, provider=provider, shard_noun=shard_noun)
    key = catalog_cache_key(path, files)
    cached = cache.get(key)
    if cached is not None:
        return cached  # type: ignore[no-any-return]
    parsed = parse(files)
    cache[key] = parsed
    return parsed

Strict YAML

The duplicate-key-rejecting loader every catalog parses through — a mapping that declares the same key twice raises ValueError rather than silently keeping the last value.

earthlens.base.yaml_loader.load_yaml_strict(path)

Parse a YAML file, rejecting duplicate mapping keys.

A thin wrapper over yaml.load(..., Loader=_StrictSafeLoader) so callers (the catalog loaders) never touch the loader class directly.

Parameters:

Name	Type	Description	Default
path	str | Path	Filesystem path to the YAML file.	required

Returns:

Type	Description
Any	The parsed YAML (typically a dict), or None for an empty
Any	file.

Raises:

Type	Description
ValueError	If any mapping in the file declares a key twice.

Examples:

• Parse a small YAML file and read a value:

  >>> import os, tempfile, textwrap
  >>> p = os.path.join(tempfile.mkdtemp(), "ok.yaml")
  >>> _ = open(p, "w").write(textwrap.dedent('''
  ...     name: demo
  ...     items:
  ...       - a
  ...       - b
  ... '''))
  >>> data = load_yaml_strict(p)
  >>> data["name"]
  'demo'
  >>> data["items"]
  ['a', 'b']
  

• A duplicate mapping key is rejected at parse time:

  >>> import os, tempfile, textwrap
  >>> p = os.path.join(tempfile.mkdtemp(), "dup.yaml")
  >>> _ = open(p, "w").write(textwrap.dedent('''
  ...     a: 1
  ...     a: 2
  ... '''))
  >>> load_yaml_strict(p)  # doctest: +ELLIPSIS
  Traceback (most recent call last):
      ...
  ValueError: duplicate YAML key 'a' at line 3, ...
  

See Also

earthlens.ecmwf.catalog.Catalog: Uses this to load the CDS catalog. earthlens.gee.catalog.Catalog: Uses this to load the GEE catalog.

Source code in libs/core/src/earthlens/base/yaml_loader.py

def load_yaml_strict(path: str | Path) -> Any:
    """Parse a YAML file, rejecting duplicate mapping keys.

    A thin wrapper over `yaml.load(..., Loader=_StrictSafeLoader)` so
    callers (the catalog loaders) never touch the loader class directly.

    Args:
        path: Filesystem path to the YAML file.

    Returns:
        The parsed YAML (typically a `dict`), or `None` for an empty
        file.

    Raises:
        ValueError: If any mapping in the file declares a key twice.

    Examples:
        - Parse a small YAML file and read a value:
            ```python
            >>> import os, tempfile, textwrap
            >>> p = os.path.join(tempfile.mkdtemp(), "ok.yaml")
            >>> _ = open(p, "w").write(textwrap.dedent('''
            ...     name: demo
            ...     items:
            ...       - a
            ...       - b
            ... '''))
            >>> data = load_yaml_strict(p)
            >>> data["name"]
            'demo'
            >>> data["items"]
            ['a', 'b']

            ```
        - A duplicate mapping key is rejected at parse time:
            ```python
            >>> import os, tempfile, textwrap
            >>> p = os.path.join(tempfile.mkdtemp(), "dup.yaml")
            >>> _ = open(p, "w").write(textwrap.dedent('''
            ...     a: 1
            ...     a: 2
            ... '''))
            >>> load_yaml_strict(p)  # doctest: +ELLIPSIS
            Traceback (most recent call last):
                ...
            ValueError: duplicate YAML key 'a' at line 3, ...

            ```

    See Also:
        earthlens.ecmwf.catalog.Catalog: Uses this to load the CDS catalog.
        earthlens.gee.catalog.Catalog: Uses this to load the GEE catalog.

    """
    with open(path, encoding="utf-8") as stream:
        # `_StrictSafeLoader` subclasses `yaml.SafeLoader` (no arbitrary
        # object instantiation); bandit's B506 flags any `yaml.load`.
        return yaml.load(stream, Loader=_StrictSafeLoader)  # nosec B506

Provider registry

Backends that populate the base providers field load it from a per-backend providers.yaml.

earthlens.base.Provider

Bases: BaseModel

One canonical data provider — a slug-id with a display name and parent.

Frozen value object loaded from a backend's providers.yaml. Datasets reference providers by slug via their provider: field; the catalog loader validates that every referenced slug is registered.

Attributes:

Name	Type	Description
slug	str	Stable kebab-case identifier (e.g. "nasa-lp-daac", "copernicus-marine", "ucsb-chc"); injected from the YAML mapping key.
display_name	str	Human-readable name to render in docs and UIs.
parent	str | None	Slug of the parent provider, or None for top-level organisations. Used to group e.g. all NASA DAACs under the "nasa" umbrella.

Source code in libs/core/src/earthlens/base/providers.py

class Provider(BaseModel):
    """One canonical data provider — a slug-id with a display name and parent.

    Frozen value object loaded from a backend's `providers.yaml`.
    Datasets reference providers by slug via their `provider:` field;
    the catalog loader validates that every referenced slug is
    registered.

    Attributes:
        slug: Stable kebab-case identifier (e.g. `"nasa-lp-daac"`,
            `"copernicus-marine"`, `"ucsb-chc"`); injected from the
            YAML mapping key.
        display_name: Human-readable name to render in docs and UIs.
        parent: Slug of the parent provider, or `None` for top-level
            organisations. Used to group e.g. all NASA DAACs under
            the `"nasa"` umbrella.
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    slug: str
    display_name: str
    parent: str | None = None

earthlens.base.load_providers(path)

Parse + cache providers.yaml at path, keyed on (path, mtime_ns).

Parameters:

Name	Type	Description	Default
path	Path	Filesystem path of a providers.yaml-shaped file (a top-level providers: map of slug -> {display_name, parent?}).	required

Returns:

Type	Description
dict[str, Provider]	slug -> Provider mapping.

Raises:

Type	Description
ValueError	If the file is missing, declares a slug whose parent is not itself a registered slug, or fails pydantic validation on any entry.

Source code in libs/core/src/earthlens/base/providers.py

def load_providers(path: Path) -> dict[str, Provider]:
    """Parse + cache `providers.yaml` at `path`, keyed on `(path, mtime_ns)`.

    Args:
        path: Filesystem path of a `providers.yaml`-shaped file (a
            top-level `providers:` map of `slug -> {display_name,
            parent?}`).

    Returns:
        `slug -> Provider` mapping.

    Raises:
        ValueError: If the file is missing, declares a slug whose
            `parent` is not itself a registered slug, or fails
            pydantic validation on any entry.
    """
    resolved = str(path.resolve())
    try:
        mtime_ns = path.stat().st_mtime_ns
    except FileNotFoundError as exc:
        raise ValueError(
            f"providers registry not found at {path}; L2 (provider "
            "normalisation) expects this file alongside the catalog."
        ) from exc
    key = (resolved, mtime_ns)
    cached = _PROVIDERS_CACHE.get(key)
    if cached is not None:
        return cached

    data = load_yaml_strict(path) or {}
    raw = data.get("providers") or {}
    out: dict[str, Provider] = {}
    for slug, body in raw.items():
        try:
            out[slug] = Provider(slug=slug, **dict(body or {}))
        except ValidationError as exc:
            raise ValueError(f"invalid provider {slug!r} in {path}: {exc}") from exc
    for slug, p in out.items():
        if p.parent is not None and p.parent not in out:
            raise ValueError(
                f"provider {slug!r} declares parent={p.parent!r}, "
                f"which is not a known provider slug in {path}"
            )
    _PROVIDERS_CACHE[key] = out
    return out

Filesystem helpers

earthlens.base.safe_filename(value)

Sanitise an id into a filesystem-safe file stem.

Replaces every maximal run of characters outside the whitelist (A-Z a-z 0-9 . _ -) with a single _, then strips any leading / trailing _. Dots are kept, so a dataset id like cmems_mod_glo_phy_my_0.083deg_P1D-m is returned unchanged while a path-bearing key like planetary-computer/sentinel-2-l2a flattens to planetary-computer_sentinel-2-l2a.

Parameters:

Name	Type	Description	Default
value	str	The raw provider id / key.	required

Returns:

Type	Description
str	A filesystem-safe stem: only A-Z a-z 0-9 . _ -, no leading /
str	trailing _.

Examples:

• Path separators and Windows-illegal characters collapse to _, while dots and hyphens survive:

  >>> from earthlens.base.naming import safe_filename
  >>> safe_filename("a/b\\c:d")
  'a_b_c_d'
  >>> safe_filename('a*b?c"d<e>f|g')
  'a_b_c_d_e_f_g'
  >>> safe_filename("cmems_mod_glo_phy_my_0.083deg_P1D-m")
  'cmems_mod_glo_phy_my_0.083deg_P1D-m'
  

Source code in libs/core/src/earthlens/base/naming.py

def safe_filename(value: str) -> str:
    r"""Sanitise an id into a filesystem-safe file stem.

    Replaces every maximal run of characters outside the whitelist
    (`A-Z a-z 0-9 . _ -`) with a single `_`, then strips any leading /
    trailing `_`. Dots are kept, so a dataset id like
    `cmems_mod_glo_phy_my_0.083deg_P1D-m` is returned unchanged while a
    path-bearing key like `planetary-computer/sentinel-2-l2a` flattens to
    `planetary-computer_sentinel-2-l2a`.

    Args:
        value: The raw provider id / key.

    Returns:
        A filesystem-safe stem: only `A-Z a-z 0-9 . _ -`, no leading /
        trailing `_`.

    Examples:
        - Path separators and Windows-illegal characters collapse to `_`,
          while dots and hyphens survive:
            ```python
            >>> from earthlens.base.naming import safe_filename
            >>> safe_filename("a/b\\c:d")
            'a_b_c_d'
            >>> safe_filename('a*b?c"d<e>f|g')
            'a_b_c_d_e_f_g'
            >>> safe_filename("cmems_mod_glo_phy_my_0.083deg_P1D-m")
            'cmems_mod_glo_phy_my_0.083deg_P1D-m'

            ```
    """
    return _UNSAFE.sub("_", value).strip("_")