fetch

def fetch(self, fname: str, offline: bool | None = None) -> Path:
    """Return a verified local path for ``fname``, downloading if needed."""
    if fname not in self.registry:
        raise KeyError(f'{fname!r} is not in the registry for {self.subdir!r}')
    known_hash = self.registry[fname]
    target = self.target_dir / fname

    if target.is_file() and _hash_matches(target, known_hash):
        self._sources.setdefault(fname, 'local')
        return target
    if target.is_file():
        log.warning('checksum mismatch for %s; refetching', target)

    cached = self._fetch_from_cache(fname, known_hash, target)
    if cached is not None:
        return cached

    if is_offline() if offline is None else offline:
        raise OfflineDataError(
            f'{target} is missing or invalid and offline mode is active; '
            f'populate the data tree at {self.data_root} or unset FWL_IO_OFFLINE'
        )
    return self._download(fname, known_hash, target)

def fetch_all(self, offline: bool | None = None) -> list[Path]:
    """Fetch every registry file, then stamp the version directory.

    The stamp is written here (the whole-dataset operation), not by an
    individual :meth:`fetch`, so a version directory populated one file at
    a time is not stamped until a ``fetch_all`` completes it. A stamp-write
    failure never fails the fetch: the data is already in place.
    """
    paths = [self.fetch(name, offline=offline) for name in sorted(self.registry)]
    self._write_stamp()
    return paths

def provenance(self) -> list[dict[str, str]]:
    """Return (file, source, checksum) records for run-provenance manifests.

    ``source`` is the actual origin of files resolved by this Fetcher
    instance (``local``, ``cache:<path>``, or the mirror that served the
    download). Files not yet fetched in this session are marked with a
    ``declared:`` prefix on the primary mirror, since their true origin
    is unknown.
    """
    return [
        {
            'file': f'{self.rel_dir}/{name}',
            'source': self._sources.get(name, f'declared:{self.mirrors[0]}'),
            'checksum': digest,
        }
        for name, digest in sorted(self.registry.items())
    ]