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

    # Fast path: an already-present, valid file needs no work and no lock,
    # so the common case (data already on disk) pays nothing for the guard.
    if target.is_file() and _hash_matches(target, known_hash):
        self._sources.setdefault(fname, 'local')
        return target

    # Serialise concurrent fetchers of the *same* file across processes: a
    # burst of instances that all miss it would otherwise each hit the
    # mirrors at once and risk getting the collaboration rate-limited. The
    # lock is per-target, so unrelated files still fetch in parallel.
    # Double-checked: re-test the target under the lock, because another
    # process may have completed the download while we waited.
    with self._fetch_lock(fname, target):
        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. An archive
    dataset is downloaded, verified, and extracted as one operation.
    """
    if self.extract is not None:
        return self._fetch_archive(offline=offline)
    paths = [self.fetch(name, offline=offline) for name in sorted(self.registry)]
    self._write_stamp()
    return paths

def file_matches(self, fname: str) -> bool:
    """True when the local file for ``fname`` matches its registry digest.

    Reads the file; it does not fetch, and it does not check that the file
    exists first, so a caller wanting to tell an absent file from a corrupt
    one tests for presence itself. An unreadable file raises ``OSError``
    rather than reporting a mismatch, since a permission problem on the
    tree is a different fault from wrong contents.

    Raises
    ------
    KeyError
        ``fname`` is not declared in this dataset's registry, so there is
        no digest to compare it against.
    OSError
        The file could not be read, whether because it is absent or
        because the tree denies access to it.
    """
    if fname not in self.registry:
        raise KeyError(f'{fname!r} is not in the registry for {self.subdir!r}')
    return _hash_matches(self.target_dir / fname, self.registry[fname])

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())
    ]

def recorded_members(self) -> list[str] | None:
    """Return the extracted members this dataset's stamp records.

    ``None`` when there is no stamp describing this dataset's tree, which
    says the tree is not there to be examined rather than that it is empty.
    Callers must keep those apart: an empty list would read as a complete
    tree of no files.

    A stamp qualifies only when it describes this deposit and this archive
    kind. One left by a different fetch of the same subdirectory, a plain
    fetch or another version, does not describe this tree.

    Members that would resolve outside the dataset directory are dropped.
    A stamp is a file on disk like any other and can be edited or replaced,
    so a name in it is treated with the same suspicion as a name inside an
    archive rather than joined onto the tree unchecked.
    """
    return self._stamp_members(self.target_dir)