Catalogue

envlib.catalogue.Catalogue

RCG-backed catalogue of envlib datasets.

Catalogue() connects to the public envlib RCG (read-only). Pass remotes=[...] (S3Connection | dict | URL str) to use your own RCGs — this replaces the public default unless include_public=True.

The catalogue snapshots all entries at construction; call :meth:refresh to re-pull after new registrations. When a remote is unreachable and a previously pulled local index exists, the catalogue degrades to the cached copy (with a warning) instead of failing.

Example

cat = Catalogue(remotes=['https://s3.example.com/bucket/catalogue.rcg']) cat.variables ['streamflow', 'temperature'] refs = cat.query(variable='temperature', feature='atmosphere') ds = refs[0].open()

Source code in envlib/catalogue.py

class Catalogue:
    """RCG-backed catalogue of envlib datasets.

    ``Catalogue()`` connects to the public envlib RCG (read-only). Pass
    ``remotes=[...]`` (S3Connection | dict | URL str) to use your own RCGs —
    this replaces the public default unless ``include_public=True``.

    The catalogue snapshots all entries at construction; call :meth:`refresh`
    to re-pull after new registrations. When a remote is unreachable and a
    previously pulled local index exists, the catalogue degrades to the cached
    copy (with a warning) instead of failing.

    Example:
        >>> cat = Catalogue(remotes=['https://s3.example.com/bucket/catalogue.rcg'])
        >>> cat.variables
        ['streamflow', 'temperature']
        >>> refs = cat.query(variable='temperature', feature='atmosphere')
        >>> ds = refs[0].open()
    """

    def __init__(self, remotes=None, cache: str = DEFAULT_CACHE_DIR, *, include_public: bool = False):
        self._cache_dir = pathlib.Path(cache).expanduser()
        self._cache_dir.mkdir(parents=True, exist_ok=True)

        sources = []
        if remotes is None:
            public = _public_rcg_url()
            if public is None:
                msg = (
                    'no public envlib RCG is configured — pass remotes=[...] '
                    f'or set the {PUBLIC_RCG_ENV_VAR} environment variable.'
                )
                raise ValueError(msg)
            sources.append(public)
        else:
            if isinstance(remotes, (ebooklet.S3Connection, str, dict)):
                remotes = [remotes]
            sources.extend(remotes)
            if include_public:
                public = _public_rcg_url()
                if public is None:
                    msg = f'include_public=True but no public RCG is configured ({PUBLIC_RCG_ENV_VAR}).'
                    raise ValueError(msg)
                sources.append(public)
        self._sources = sources
        self._entries: dict = {}
        self.refresh()

    # -- index management ----------------------------------------------------

    def _rcg_cache_path(self, source) -> pathlib.Path:
        self._cache_dir.mkdir(parents=True, exist_ok=True)
        return self._cache_dir / f'{_conn_cache_key(source)}.rcg'

    def refresh(self):
        """Re-pull the RCG index from all configured remotes."""
        entries: dict = {}
        for source in self._sources:
            path = self._rcg_cache_path(source)
            try:
                # offline='auto': ebooklet falls back to the cached local index
                # (with its own UserWarning) ONLY on transport-level
                # unreachability. Everything else raises typed: a broken store
                # (RemoteIntegrityError), an incompatible format
                # (UnsupportedFormatError - pre-0.10 this was silently
                # mislabeled "not readable yet" by a blanket ValueError catch),
                # and unreachable-with-no-cache (OfflineError) all propagate.
                with ebooklet.open_rcg(source, path, flag='r', offline='auto') as rcg:
                    source_entries = {k: v for k, v in rcg.items() if _HEX24_RE.fullmatch(str(k))}
            except ebooklet.UUIDMismatchError as err:
                # The local cache identifies a DIFFERENT database than the
                # remote - the remote was likely deleted and recreated.
                # Distinct from the bootstrap case: the fix is deleting the
                # stale cache file, so say so instead of a generic warning.
                warnings.warn(
                    f'RCG source identity mismatch ({err}); the remote was likely deleted and '
                    f'recreated. Delete the stale cache at {path} to adopt the new remote. '
                    'Treating this source as empty for now.',
                    stacklevel=2,
                )
                source_entries = {}
            except ebooklet.RemoteMissingError as err:
                # The RCG does not exist on the remote yet (and no local copy
                # exists) — the bootstrap case: a producer constructs the
                # Catalogue before the first publish creates the RCG. Treat as
                # an empty source, loudly.
                warnings.warn(f'RCG source not readable yet ({err}); treating as empty.', stacklevel=2)
                source_entries = {}
            for key, entry in source_entries.items():
                entries.setdefault(key, entry)  # first configured source wins on duplicates
        self._entries = entries

    @property
    def datasets(self) -> list:
        """All catalogue entries as DatasetRef objects."""
        return [DatasetRef(key, entry, self._cache_dir) for key, entry in self._entries.items()]

    # -- value discovery -------------------------------------------------------

    def distinct(self, field: str, *, counts: bool = False):
        """Distinct stored values of a queryable field across the catalogue.

        The browse companion to :meth:`query`: ``cat.distinct('variable')``
        tells you what is actually *in* this catalogue (the vocabularies module
        lists what is *valid*, and can't help at all for the free-form fields
        ``owner``/``product_code``/``version``).

        Args:
            field: Any queryable field (identity fields, ``license``,
                ``dataset_type``, ...).
            counts: When False (default), return a sorted list of the distinct
                values, excluding None. When True, return a ``{value: count}``
                dict that DOES include a None key when some entries lack the
                field (e.g. datasets with no ``product_code``).
        """
        if field not in _QUERYABLE_FIELDS:
            msg = f'unknown field {field!r}; queryable fields are {sorted(_QUERYABLE_FIELDS)}.'
            raise ValueError(msg)
        tally: dict = {}
        for entry in self._entries.values():
            value = (entry.get('user_meta') or {}).get(field)
            tally[value] = tally.get(value, 0) + 1
        if counts:
            return dict(sorted(tally.items(), key=lambda kv: (kv[0] is None, str(kv[0]))))
        return sorted(v for v in tally if v is not None)

    @property
    def features(self) -> list:
        """Distinct feature values present in the catalogue (sorted)."""
        return self.distinct('feature')

    @property
    def variables(self) -> list:
        """Distinct variable values present in the catalogue (sorted)."""
        return self.distinct('variable')

    @property
    def methods(self) -> list:
        """Distinct method values present in the catalogue (sorted)."""
        return self.distinct('method')

    @property
    def product_codes(self) -> list:
        """Distinct product_code values present in the catalogue (sorted; None excluded — see distinct())."""
        return self.distinct('product_code')

    @property
    def processing_levels(self) -> list:
        """Distinct processing_level values present in the catalogue (sorted)."""
        return self.distinct('processing_level')

    @property
    def owners(self) -> list:
        """Distinct owner values present in the catalogue (sorted)."""
        return self.distinct('owner')

    @property
    def aggregation_statistics(self) -> list:
        """Distinct aggregation_statistic values present in the catalogue (sorted)."""
        return self.distinct('aggregation_statistic')

    @property
    def frequency_intervals(self) -> list:
        """Distinct frequency_interval codes present in the catalogue (sorted; None excluded)."""
        return self.distinct('frequency_interval')

    @property
    def utc_offsets(self) -> list:
        """Distinct utc_offset values present in the catalogue (sorted)."""
        return self.distinct('utc_offset')

    @property
    def spatial_resolutions(self) -> list:
        """Distinct spatial_resolution values present in the catalogue (sorted; None excluded)."""
        return self.distinct('spatial_resolution')

    @property
    def versions(self) -> list:
        """Distinct version strings present in the catalogue (sorted).

        Version spellings are per-dataset conventions, so this global list
        mixes unrelated series — mostly useful after narrowing with query().
        """
        return self.distinct('version')

    @property
    def licenses(self) -> list:
        """Distinct license values present in the catalogue (sorted)."""
        return self.distinct('license')

    @property
    def dataset_types(self) -> list:
        """Distinct dataset_type values present in the catalogue (sorted)."""
        return self.distinct('dataset_type')

    # -- query ----------------------------------------------------------------

    def query(
        self,
        *,
        bbox=None,
        within_radius=None,
        geometry=None,
        start_date=None,
        end_date=None,
        **fields,
    ) -> list:
        """Filter the catalogue; kwargs are AND'd, a list value means any-of.

        Spatial filters (mutually exclusive, EPSG:4326): ``bbox`` (intersects),
        ``within_radius`` (((lon, lat), km) great-circle), ``geometry``
        (shapely, intersects). Temporal: ``start_date``/``end_date`` overlap
        the dataset's time range. Without an explicit ``version=`` kwarg the
        latest version (greatest created_at) of each matching dataset is
        returned.
        """
        unknown = set(fields) - _QUERYABLE_FIELDS
        if unknown:
            msg = f'unknown query fields: {sorted(unknown)}; queryable fields are {sorted(_QUERYABLE_FIELDS)}.'
            raise ValueError(msg)
        spatial_kwargs = (('bbox', bbox), ('within_radius', within_radius), ('geometry', geometry))
        spatial = [kw for kw, v in spatial_kwargs if v is not None]
        if len(spatial) > 1:
            msg = f'spatial filters are mutually exclusive; got {spatial}.'
            raise ValueError(msg)

        results = []
        for ref in self.datasets:
            user_meta = ref.metadata
            if not all(_field_matches(user_meta.get(name), wanted) for name, wanted in fields.items()):
                continue
            stored_bbox = user_meta.get('bbox')
            if bbox is not None and (stored_bbox is None or not _bbox_intersects(stored_bbox, bbox)):
                continue
            if geometry is not None and (stored_bbox is None or not _geometry_intersects(stored_bbox, geometry)):
                continue
            if within_radius is not None:
                (lon, lat), radius_km = within_radius
                if stored_bbox is None or not _bbox_within_radius(stored_bbox, lon, lat, radius_km):
                    continue
            if (start_date is not None or end_date is not None) and not _time_overlaps(
                user_meta.get('time_start'), user_meta.get('time_end'), start_date, end_date
            ):
                continue
            results.append(ref)

        if 'version' not in fields:
            results = _latest_per_dataset(results)
        return results

    # -- validate / publish / register / deregister ---------------------------

    def validate(self, local_cfdb_path) -> dict:
        """Validate a local cfdb file against envlib's requirements (no RCG or S3 changes).

        Returns a summary dict ({'metadata', 'dataset_version_id', 'dataset_id', 'state',
        'standard_name'}); raises ValidationError on invalid input.

        Thin wrapper over the module-level :func:`validate_dataset`, which needs no Catalogue
        and no network -- prefer that one when you are not publishing.
        """
        return validate_dataset(local_cfdb_path)

    def publish(
        self, local_cfdb_path, remote_conn, rcg_remote_conn, num_groups=None, verify_objects: bool = True, **open_kwargs
    ) -> dict:
        """Validate, push the cfdb data to its S3 remote, verify it, then register it in the RCG.

        The cfdb data is pushed BEFORE the RCG entry so the catalogue never
        references incomplete remote data. With ``verify_objects`` (default True),
        after the push and before the entry is written the remote is fsck'd, and a
        silent over-claim (the committed index references objects the store does not
        actually hold) raises ``PublishIntegrityError`` so a broken dataset is never
        advertised. A re-run after a LOUD push failure is safe (idempotent push,
        upsert entry); a re-run after a SILENT over-claim does NOT heal it - recover
        by retract + full republish (see ``_verify_remote_objects``).
        """
        member_conn = _as_connection(remote_conn)
        edataset_kwargs = dict(open_kwargs)
        if num_groups is not None:
            edataset_kwargs['num_groups'] = num_groups
        # validate INSIDE the edataset session: for a re-publish of an
        # already-pushed (possibly partially materialized) local file, plain
        # open_dataset would read local chunks only and could extract wrong
        # extents; the EDataset pulls transparently. A ValidationError aborts
        # before anything is pushed.
        with cfdb.open_edataset(member_conn, local_cfdb_path, flag='w', **edataset_kwargs) as eds:
            # attrs carrying a dataset_version_id mean this dataset was validated/registered
            # before: skip CV re-validation (validation on change only).
            first_time = eds.attrs.get('envlib_dataset_version_id') is None
            result = _validate_dataset(eds, validate_cv=first_time)
            _apply_derived_attrs(eds, result)
            _raise_on_push_failure(eds.push(), 'publish: pushing the dataset data')

        if verify_objects:
            _verify_remote_objects(member_conn)
        self._upsert_entry(rcg_remote_conn, member_conn, result)
        return result

    def register(self, remote_conn, rcg_remote_conn, verify_objects: bool = True, **open_kwargs) -> dict:
        """Register an already-remote cfdb file in the catalogue (no data push).

        ``remote_conn`` must be writable (credentials): first registration
        writes the self-identification attrs (and any auto-populated
        standard_name) into the dataset, pushing that metadata-only change.
        """
        member_conn = _as_connection(remote_conn)
        local_path = self._cache_dir / f'{_conn_cache_key(member_conn)}.cfdb'
        with cfdb.open_edataset(member_conn, local_path, flag='w', **open_kwargs) as eds:
            first_time = eds.attrs.get('envlib_dataset_version_id') is None
            result = _validate_dataset(eds, validate_cv=first_time)
            if _apply_derived_attrs(eds, result):
                _raise_on_push_failure(eds.push(), 'register: pushing the self-identification attrs')

        if verify_objects:
            _verify_remote_objects(member_conn)
        self._upsert_entry(rcg_remote_conn, member_conn, result)
        return result

    def deregister(
        self,
        dataset_version_id: str,
        rcg_remote_conn,
        *,
        delete_data: bool = False,
        access_key_id=None,
        access_key=None,
    ):
        """Remove a dataset's catalogue entry; optionally delete the hosted data.

        Plain deregistration only delists — the hosted data stays up for
        existing consumers. ``delete_data=True`` (retraction) additionally
        deletes the remote cfdb via ebooklet, after verifying no OTHER entry
        references the same remote target (the shared-target guard); it needs
        the data owner's credentials injected.
        """
        rcg_conn = _as_connection(rcg_remote_conn)
        with ebooklet.open_rcg(rcg_conn, self._rcg_cache_path(rcg_conn), flag='w') as rcg:
            entry = rcg.get(dataset_version_id)
            if entry is None:
                msg = f'no catalogue entry for dataset_version_id {dataset_version_id!r}.'
                raise ValidationError(msg)
            if delete_data:
                target_conn = entry.get('remote_conn') or {}
                target = (target_conn.get('endpoint_url'), target_conn.get('bucket'), target_conn.get('db_key'))
                # list() first: fetching entries while iterating keys() would
                # deadlock on the underlying booklet thread lock.
                for other_key in list(rcg.keys()):
                    if other_key == dataset_version_id or not _HEX24_RE.fullmatch(str(other_key)):
                        continue
                    other = rcg.get(other_key) or {}
                    other_conn = other.get('remote_conn') or {}
                    if (other_conn.get('endpoint_url'), other_conn.get('bucket'), other_conn.get('db_key')) == target:
                        msg = (
                            f'refusing delete_data=True: entry {other_key!r} references the same remote '
                            f'target ({target[2]!r} in bucket {target[1]!r}); deleting would destroy its data. '
                            f'Deregister without delete_data, or resolve the shared target first.'
                        )
                        raise ValidationError(msg)
                if access_key_id is None or access_key is None:
                    msg = 'delete_data=True needs the data owner credentials (access_key_id/access_key).'
                    raise ValidationError(msg)
                member_conn = ebooklet.S3Connection(
                    access_key_id=access_key_id,
                    access_key=access_key,
                    db_key=target_conn.get('db_key'),
                    bucket=target_conn.get('bucket'),
                    endpoint_url=target_conn.get('endpoint_url'),
                )
                with member_conn.open('w') as session:
                    session.delete_remote()
            del rcg[dataset_version_id]
            _raise_on_push_failure(rcg.changes().push(), 'deregister: pushing the catalogue entry removal')
        self.refresh()

    # -- entry construction ----------------------------------------------------

    def _upsert_entry(self, rcg_remote_conn, member_conn: ebooklet.S3Connection, result: dict):
        rcg_conn = _as_connection(rcg_remote_conn)
        dataset_version_id = result['dataset_version_id']
        with ebooklet.open_rcg(rcg_conn, self._rcg_cache_path(rcg_conn), flag='c') as rcg:
            existing = rcg.get(dataset_version_id)
            existing_meta = (existing or {}).get('user_meta') or {}
            ## dataset_type is not an identity field, so two datasets differing ONLY by type
            ## hash to the same key and this upsert would silently replace one with the other
            ## -- bbox, time range and all. Reachable today with a plain grid and ts_ortho
            ## pair, not just with the forecast types. Refuse instead: a genuine type change
            ## is a different dataset and should be deregistered first.
            existing_type = existing_meta.get('dataset_type')
            incoming_type = result['state'].get('dataset_type')
            if existing is not None and existing_type and existing_type != incoming_type:
                msg = (
                    f'catalogue entry {dataset_version_id} is already registered with '
                    f'dataset_type {existing_type!r}, but this dataset is {incoming_type!r}. '
                    f'These share a dataset_version_id because dataset_type is not an identity '
                    f'field. Deregister the existing entry first, or change an identity field '
                    f'(method, product_code or version) so the two are distinct.'
                )
                raise ValidationError(msg)

            now = _utc_now_iso()
            user_meta = _build_user_meta(result, member_conn)
            user_meta['created_at'] = existing_meta.get('created_at') or now

            comparable_new = {k: v for k, v in user_meta.items() if k != 'modified_at'}
            comparable_old = {k: v for k, v in existing_meta.items() if k != 'modified_at'}
            stored_conn = (existing or {}).get('remote_conn') or {}
            conn_changed = stored_conn != member_conn.to_dict()
            if existing is not None and comparable_new == comparable_old and not conn_changed:
                return  # true no-op: do not bump modified_at, do not push

            user_meta['modified_at'] = now
            rcg.add(member_conn, key=dataset_version_id, user_meta=user_meta)
            _raise_on_push_failure(rcg.changes().push(), 'publish/register: pushing the catalogue entry')
        self.refresh()

aggregation_statistics property

Distinct aggregation_statistic values present in the catalogue (sorted).

dataset_types property

Distinct dataset_type values present in the catalogue (sorted).

datasets property

All catalogue entries as DatasetRef objects.

features property

Distinct feature values present in the catalogue (sorted).

frequency_intervals property

Distinct frequency_interval codes present in the catalogue (sorted; None excluded).

licenses property

Distinct license values present in the catalogue (sorted).

methods property

Distinct method values present in the catalogue (sorted).

owners property

Distinct owner values present in the catalogue (sorted).

processing_levels property

Distinct processing_level values present in the catalogue (sorted).

product_codes property

Distinct product_code values present in the catalogue (sorted; None excluded — see distinct()).

spatial_resolutions property

Distinct spatial_resolution values present in the catalogue (sorted; None excluded).

utc_offsets property

Distinct utc_offset values present in the catalogue (sorted).

variables property

Distinct variable values present in the catalogue (sorted).

versions property

Distinct version strings present in the catalogue (sorted).

Version spellings are per-dataset conventions, so this global list mixes unrelated series — mostly useful after narrowing with query().

deregister(dataset_version_id, rcg_remote_conn, *, delete_data=False, access_key_id=None, access_key=None)

Remove a dataset's catalogue entry; optionally delete the hosted data.

Plain deregistration only delists — the hosted data stays up for existing consumers. delete_data=True (retraction) additionally deletes the remote cfdb via ebooklet, after verifying no OTHER entry references the same remote target (the shared-target guard); it needs the data owner's credentials injected.

Source code in envlib/catalogue.py

def deregister(
    self,
    dataset_version_id: str,
    rcg_remote_conn,
    *,
    delete_data: bool = False,
    access_key_id=None,
    access_key=None,
):
    """Remove a dataset's catalogue entry; optionally delete the hosted data.

    Plain deregistration only delists — the hosted data stays up for
    existing consumers. ``delete_data=True`` (retraction) additionally
    deletes the remote cfdb via ebooklet, after verifying no OTHER entry
    references the same remote target (the shared-target guard); it needs
    the data owner's credentials injected.
    """
    rcg_conn = _as_connection(rcg_remote_conn)
    with ebooklet.open_rcg(rcg_conn, self._rcg_cache_path(rcg_conn), flag='w') as rcg:
        entry = rcg.get(dataset_version_id)
        if entry is None:
            msg = f'no catalogue entry for dataset_version_id {dataset_version_id!r}.'
            raise ValidationError(msg)
        if delete_data:
            target_conn = entry.get('remote_conn') or {}
            target = (target_conn.get('endpoint_url'), target_conn.get('bucket'), target_conn.get('db_key'))
            # list() first: fetching entries while iterating keys() would
            # deadlock on the underlying booklet thread lock.
            for other_key in list(rcg.keys()):
                if other_key == dataset_version_id or not _HEX24_RE.fullmatch(str(other_key)):
                    continue
                other = rcg.get(other_key) or {}
                other_conn = other.get('remote_conn') or {}
                if (other_conn.get('endpoint_url'), other_conn.get('bucket'), other_conn.get('db_key')) == target:
                    msg = (
                        f'refusing delete_data=True: entry {other_key!r} references the same remote '
                        f'target ({target[2]!r} in bucket {target[1]!r}); deleting would destroy its data. '
                        f'Deregister without delete_data, or resolve the shared target first.'
                    )
                    raise ValidationError(msg)
            if access_key_id is None or access_key is None:
                msg = 'delete_data=True needs the data owner credentials (access_key_id/access_key).'
                raise ValidationError(msg)
            member_conn = ebooklet.S3Connection(
                access_key_id=access_key_id,
                access_key=access_key,
                db_key=target_conn.get('db_key'),
                bucket=target_conn.get('bucket'),
                endpoint_url=target_conn.get('endpoint_url'),
            )
            with member_conn.open('w') as session:
                session.delete_remote()
        del rcg[dataset_version_id]
        _raise_on_push_failure(rcg.changes().push(), 'deregister: pushing the catalogue entry removal')
    self.refresh()

distinct(field, *, counts=False)

Distinct stored values of a queryable field across the catalogue.

The browse companion to :meth:query: cat.distinct('variable') tells you what is actually in this catalogue (the vocabularies module lists what is valid, and can't help at all for the free-form fields owner/product_code/version).

Parameters:

Name	Type	Description	Default
field	str	Any queryable field (identity fields, license, dataset_type, ...).	required
counts	bool	When False (default), return a sorted list of the distinct values, excluding None. When True, return a {value: count} dict that DOES include a None key when some entries lack the field (e.g. datasets with no product_code).	False

Source code in envlib/catalogue.py

def distinct(self, field: str, *, counts: bool = False):
    """Distinct stored values of a queryable field across the catalogue.

    The browse companion to :meth:`query`: ``cat.distinct('variable')``
    tells you what is actually *in* this catalogue (the vocabularies module
    lists what is *valid*, and can't help at all for the free-form fields
    ``owner``/``product_code``/``version``).

    Args:
        field: Any queryable field (identity fields, ``license``,
            ``dataset_type``, ...).
        counts: When False (default), return a sorted list of the distinct
            values, excluding None. When True, return a ``{value: count}``
            dict that DOES include a None key when some entries lack the
            field (e.g. datasets with no ``product_code``).
    """
    if field not in _QUERYABLE_FIELDS:
        msg = f'unknown field {field!r}; queryable fields are {sorted(_QUERYABLE_FIELDS)}.'
        raise ValueError(msg)
    tally: dict = {}
    for entry in self._entries.values():
        value = (entry.get('user_meta') or {}).get(field)
        tally[value] = tally.get(value, 0) + 1
    if counts:
        return dict(sorted(tally.items(), key=lambda kv: (kv[0] is None, str(kv[0]))))
    return sorted(v for v in tally if v is not None)

publish(local_cfdb_path, remote_conn, rcg_remote_conn, num_groups=None, verify_objects=True, **open_kwargs)

Validate, push the cfdb data to its S3 remote, verify it, then register it in the RCG.

The cfdb data is pushed BEFORE the RCG entry so the catalogue never references incomplete remote data. With verify_objects (default True), after the push and before the entry is written the remote is fsck'd, and a silent over-claim (the committed index references objects the store does not actually hold) raises PublishIntegrityError so a broken dataset is never advertised. A re-run after a LOUD push failure is safe (idempotent push, upsert entry); a re-run after a SILENT over-claim does NOT heal it - recover by retract + full republish (see _verify_remote_objects).

Source code in envlib/catalogue.py

def publish(
    self, local_cfdb_path, remote_conn, rcg_remote_conn, num_groups=None, verify_objects: bool = True, **open_kwargs
) -> dict:
    """Validate, push the cfdb data to its S3 remote, verify it, then register it in the RCG.

    The cfdb data is pushed BEFORE the RCG entry so the catalogue never
    references incomplete remote data. With ``verify_objects`` (default True),
    after the push and before the entry is written the remote is fsck'd, and a
    silent over-claim (the committed index references objects the store does not
    actually hold) raises ``PublishIntegrityError`` so a broken dataset is never
    advertised. A re-run after a LOUD push failure is safe (idempotent push,
    upsert entry); a re-run after a SILENT over-claim does NOT heal it - recover
    by retract + full republish (see ``_verify_remote_objects``).
    """
    member_conn = _as_connection(remote_conn)
    edataset_kwargs = dict(open_kwargs)
    if num_groups is not None:
        edataset_kwargs['num_groups'] = num_groups
    # validate INSIDE the edataset session: for a re-publish of an
    # already-pushed (possibly partially materialized) local file, plain
    # open_dataset would read local chunks only and could extract wrong
    # extents; the EDataset pulls transparently. A ValidationError aborts
    # before anything is pushed.
    with cfdb.open_edataset(member_conn, local_cfdb_path, flag='w', **edataset_kwargs) as eds:
        # attrs carrying a dataset_version_id mean this dataset was validated/registered
        # before: skip CV re-validation (validation on change only).
        first_time = eds.attrs.get('envlib_dataset_version_id') is None
        result = _validate_dataset(eds, validate_cv=first_time)
        _apply_derived_attrs(eds, result)
        _raise_on_push_failure(eds.push(), 'publish: pushing the dataset data')

    if verify_objects:
        _verify_remote_objects(member_conn)
    self._upsert_entry(rcg_remote_conn, member_conn, result)
    return result

query(*, bbox=None, within_radius=None, geometry=None, start_date=None, end_date=None, **fields)

Filter the catalogue; kwargs are AND'd, a list value means any-of.

Spatial filters (mutually exclusive, EPSG:4326): bbox (intersects), within_radius (((lon, lat), km) great-circle), geometry (shapely, intersects). Temporal: start_date/end_date overlap the dataset's time range. Without an explicit version= kwarg the latest version (greatest created_at) of each matching dataset is returned.

Source code in envlib/catalogue.py

def query(
    self,
    *,
    bbox=None,
    within_radius=None,
    geometry=None,
    start_date=None,
    end_date=None,
    **fields,
) -> list:
    """Filter the catalogue; kwargs are AND'd, a list value means any-of.

    Spatial filters (mutually exclusive, EPSG:4326): ``bbox`` (intersects),
    ``within_radius`` (((lon, lat), km) great-circle), ``geometry``
    (shapely, intersects). Temporal: ``start_date``/``end_date`` overlap
    the dataset's time range. Without an explicit ``version=`` kwarg the
    latest version (greatest created_at) of each matching dataset is
    returned.
    """
    unknown = set(fields) - _QUERYABLE_FIELDS
    if unknown:
        msg = f'unknown query fields: {sorted(unknown)}; queryable fields are {sorted(_QUERYABLE_FIELDS)}.'
        raise ValueError(msg)
    spatial_kwargs = (('bbox', bbox), ('within_radius', within_radius), ('geometry', geometry))
    spatial = [kw for kw, v in spatial_kwargs if v is not None]
    if len(spatial) > 1:
        msg = f'spatial filters are mutually exclusive; got {spatial}.'
        raise ValueError(msg)

    results = []
    for ref in self.datasets:
        user_meta = ref.metadata
        if not all(_field_matches(user_meta.get(name), wanted) for name, wanted in fields.items()):
            continue
        stored_bbox = user_meta.get('bbox')
        if bbox is not None and (stored_bbox is None or not _bbox_intersects(stored_bbox, bbox)):
            continue
        if geometry is not None and (stored_bbox is None or not _geometry_intersects(stored_bbox, geometry)):
            continue
        if within_radius is not None:
            (lon, lat), radius_km = within_radius
            if stored_bbox is None or not _bbox_within_radius(stored_bbox, lon, lat, radius_km):
                continue
        if (start_date is not None or end_date is not None) and not _time_overlaps(
            user_meta.get('time_start'), user_meta.get('time_end'), start_date, end_date
        ):
            continue
        results.append(ref)

    if 'version' not in fields:
        results = _latest_per_dataset(results)
    return results

refresh()

Re-pull the RCG index from all configured remotes.

Source code in envlib/catalogue.py

def refresh(self):
    """Re-pull the RCG index from all configured remotes."""
    entries: dict = {}
    for source in self._sources:
        path = self._rcg_cache_path(source)
        try:
            # offline='auto': ebooklet falls back to the cached local index
            # (with its own UserWarning) ONLY on transport-level
            # unreachability. Everything else raises typed: a broken store
            # (RemoteIntegrityError), an incompatible format
            # (UnsupportedFormatError - pre-0.10 this was silently
            # mislabeled "not readable yet" by a blanket ValueError catch),
            # and unreachable-with-no-cache (OfflineError) all propagate.
            with ebooklet.open_rcg(source, path, flag='r', offline='auto') as rcg:
                source_entries = {k: v for k, v in rcg.items() if _HEX24_RE.fullmatch(str(k))}
        except ebooklet.UUIDMismatchError as err:
            # The local cache identifies a DIFFERENT database than the
            # remote - the remote was likely deleted and recreated.
            # Distinct from the bootstrap case: the fix is deleting the
            # stale cache file, so say so instead of a generic warning.
            warnings.warn(
                f'RCG source identity mismatch ({err}); the remote was likely deleted and '
                f'recreated. Delete the stale cache at {path} to adopt the new remote. '
                'Treating this source as empty for now.',
                stacklevel=2,
            )
            source_entries = {}
        except ebooklet.RemoteMissingError as err:
            # The RCG does not exist on the remote yet (and no local copy
            # exists) — the bootstrap case: a producer constructs the
            # Catalogue before the first publish creates the RCG. Treat as
            # an empty source, loudly.
            warnings.warn(f'RCG source not readable yet ({err}); treating as empty.', stacklevel=2)
            source_entries = {}
        for key, entry in source_entries.items():
            entries.setdefault(key, entry)  # first configured source wins on duplicates
    self._entries = entries

register(remote_conn, rcg_remote_conn, verify_objects=True, **open_kwargs)

Register an already-remote cfdb file in the catalogue (no data push).

remote_conn must be writable (credentials): first registration writes the self-identification attrs (and any auto-populated standard_name) into the dataset, pushing that metadata-only change.

Source code in envlib/catalogue.py

def register(self, remote_conn, rcg_remote_conn, verify_objects: bool = True, **open_kwargs) -> dict:
    """Register an already-remote cfdb file in the catalogue (no data push).

    ``remote_conn`` must be writable (credentials): first registration
    writes the self-identification attrs (and any auto-populated
    standard_name) into the dataset, pushing that metadata-only change.
    """
    member_conn = _as_connection(remote_conn)
    local_path = self._cache_dir / f'{_conn_cache_key(member_conn)}.cfdb'
    with cfdb.open_edataset(member_conn, local_path, flag='w', **open_kwargs) as eds:
        first_time = eds.attrs.get('envlib_dataset_version_id') is None
        result = _validate_dataset(eds, validate_cv=first_time)
        if _apply_derived_attrs(eds, result):
            _raise_on_push_failure(eds.push(), 'register: pushing the self-identification attrs')

    if verify_objects:
        _verify_remote_objects(member_conn)
    self._upsert_entry(rcg_remote_conn, member_conn, result)
    return result

validate(local_cfdb_path)

Validate a local cfdb file against envlib's requirements (no RCG or S3 changes).

Returns a summary dict ({'metadata', 'dataset_version_id', 'dataset_id', 'state', 'standard_name'}); raises ValidationError on invalid input.

Thin wrapper over the module-level :func:validate_dataset, which needs no Catalogue and no network -- prefer that one when you are not publishing.

Source code in envlib/catalogue.py

def validate(self, local_cfdb_path) -> dict:
    """Validate a local cfdb file against envlib's requirements (no RCG or S3 changes).

    Returns a summary dict ({'metadata', 'dataset_version_id', 'dataset_id', 'state',
    'standard_name'}); raises ValidationError on invalid input.

    Thin wrapper over the module-level :func:`validate_dataset`, which needs no Catalogue
    and no network -- prefer that one when you are not publishing.
    """
    return validate_dataset(local_cfdb_path)

envlib.catalogue.DatasetRef

A catalogue entry: one dataset version's metadata, plus how to open its cfdb file.

Source code in envlib/catalogue.py

class DatasetRef:
    """A catalogue entry: one dataset version's metadata, plus how to open its cfdb file."""

    def __init__(self, dataset_version_id: str, entry: dict, cache_dir: pathlib.Path):
        self._dataset_id = dataset_version_id
        self._entry = entry
        self._cache_dir = cache_dir

    @property
    def metadata(self) -> dict:
        """The full envlib metadata dict stored in the catalogue entry."""
        return dict(self._entry.get('user_meta') or {})

    @property
    def entry(self) -> dict:
        """The raw RCG entry (entry schema v1: remote_conn, remote_meta, user_meta, ...)."""
        return self._entry

    def __getattr__(self, name):
        if name.startswith('_'):
            raise AttributeError(name)
        user_meta = self._entry.get('user_meta') or {}
        if name in user_meta:
            return user_meta[name]
        msg = f'{type(self).__name__} has no attribute or metadata field {name!r}.'
        raise AttributeError(msg)

    def __repr__(self):
        return repr(self.metadata)

    def open(self, file_path=None, access_key_id=None, access_key=None):
        """Open this dataset version's cfdb file as a read-only EDataset.

        Entries never store credentials. Public-HTTPS datasets open via their
        ``data_url`` with no credentials; for private buckets inject
        ``access_key_id``/``access_key``. Only inject credentials for entries
        whose endpoint/bucket you trust — injected keys sign requests against
        the entry's stored endpoint.
        """
        conn_dict = dict(self._entry.get('remote_conn') or {})
        if access_key_id is not None or access_key is not None:
            conn_dict['access_key_id'] = access_key_id
            conn_dict['access_key'] = access_key
        elif not conn_dict.get('db_url'):
            msg = (
                'this entry has no public db_url; the dataset lives on a private bucket — '
                'inject access_key_id/access_key to open it.'
            )
            raise ValidationError(msg)
        else:
            # url-only read session; drop the S3 fields so no signing is attempted
            conn_dict = {'db_url': conn_dict['db_url']}
        conn = ebooklet.S3Connection(**conn_dict)
        if file_path is None:
            self._cache_dir.mkdir(parents=True, exist_ok=True)
            file_path = self._cache_dir / f'{self._dataset_id}.cfdb'
        return cfdb.open_edataset(conn, file_path, flag='r')

entry property

The raw RCG entry (entry schema v1: remote_conn, remote_meta, user_meta, ...).

metadata property

The full envlib metadata dict stored in the catalogue entry.

open(file_path=None, access_key_id=None, access_key=None)

Open this dataset version's cfdb file as a read-only EDataset.

Entries never store credentials. Public-HTTPS datasets open via their data_url with no credentials; for private buckets inject access_key_id/access_key. Only inject credentials for entries whose endpoint/bucket you trust — injected keys sign requests against the entry's stored endpoint.

Source code in envlib/catalogue.py

def open(self, file_path=None, access_key_id=None, access_key=None):
    """Open this dataset version's cfdb file as a read-only EDataset.

    Entries never store credentials. Public-HTTPS datasets open via their
    ``data_url`` with no credentials; for private buckets inject
    ``access_key_id``/``access_key``. Only inject credentials for entries
    whose endpoint/bucket you trust — injected keys sign requests against
    the entry's stored endpoint.
    """
    conn_dict = dict(self._entry.get('remote_conn') or {})
    if access_key_id is not None or access_key is not None:
        conn_dict['access_key_id'] = access_key_id
        conn_dict['access_key'] = access_key
    elif not conn_dict.get('db_url'):
        msg = (
            'this entry has no public db_url; the dataset lives on a private bucket — '
            'inject access_key_id/access_key to open it.'
        )
        raise ValidationError(msg)
    else:
        # url-only read session; drop the S3 fields so no signing is attempted
        conn_dict = {'db_url': conn_dict['db_url']}
    conn = ebooklet.S3Connection(**conn_dict)
    if file_path is None:
        self._cache_dir.mkdir(parents=True, exist_ok=True)
        file_path = self._cache_dir / f'{self._dataset_id}.cfdb'
    return cfdb.open_edataset(conn, file_path, flag='r')