From fb1dd8139331d6c193614244de377855cbf5c3ee Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Tue, 3 Nov 2026 22:21:08 +0100 Subject: [PATCH 1/5] feat(registry): copy a release without changing a published version copy_release(reference, source, dest) copies a release and each payload source that its manifest lists into another registry. A release needs exactly those releases, never the payload sources of a payload source. With with_payload_sources=False, the copy takes the release alone. Sources come first, and each release is copied once. A published version cannot change: if the destination holds the same file records, nothing happens, or a local destination gets the release again from the source. If it holds other files, TimeNetVersionConflictError stops the copy before anything is written. Every check runs before the first copy, access included: a registry that is not local serves files only for releases whose card and lineage declare access open. Between two registries that are not local, only the missing release is downloaded. find_payload_sources() in timenet.registry.dependencies applies this rule and the access check. It finds each payload source in the registry to copy from, else as a complete copy in the destination. The typed progress events live in timenet.progress. The step events carry the release and the action, so their shape stays the same while the coordinator's step type grows. Part of TNET-157. --- packages/timenet/src/timenet/errors.py | 34 ++ packages/timenet/src/timenet/progress.py | 75 +++ packages/timenet/src/timenet/registry/copy.py | 225 +++++++++ .../src/timenet/registry/dependencies.py | 71 ++- packages/timenet/tests/registry/test_copy.py | 467 ++++++++++++++++++ packages/timenet/tests/test_errors.py | 13 + packages/timenet/tests/test_progress.py | 12 + zensical.toml | 2 +- 8 files changed, 897 insertions(+), 2 deletions(-) create mode 100644 packages/timenet/src/timenet/progress.py create mode 100644 packages/timenet/src/timenet/registry/copy.py create mode 100644 packages/timenet/tests/registry/test_copy.py create mode 100644 packages/timenet/tests/test_progress.py diff --git a/packages/timenet/src/timenet/errors.py b/packages/timenet/src/timenet/errors.py index a1ee95f82..c0a198176 100644 --- a/packages/timenet/src/timenet/errors.py +++ b/packages/timenet/src/timenet/errors.py @@ -58,6 +58,40 @@ def __str__(self) -> str: return message if self.hint is None else f"{message} {self.hint}" +class TimeNetVersionConflictError(TimeNetError): + """A registry holds a release with other files than the copy that would go there. + + A published version cannot change. Other files need a new version. + + Attributes: + release: The release that the registry holds with other files. + registry: The registry that holds it, as text. + source: The registry that the copy comes from, as text. + needed_by: The release that the copy is for, if ``release`` is a payload source that it + needs, directly or through another payload source. ``None`` if ``release`` is the + release that the copy is for. + hint: What the caller can do instead, added by the front end, or ``None``. + """ + + def __init__( + self, release: "DatasetRef", registry: str, source: str, needed_by: "DatasetRef | None" = None + ) -> None: + super().__init__(release, registry, source, needed_by) + self.release = release + self.registry = registry + self.source = source + self.needed_by = needed_by + self.hint: str | None = None + + def __str__(self) -> str: + held = str(self.release) if self.needed_by is None else f"{self.release}, a payload source of {self.needed_by}," + message = ( + f"{self.registry} already holds {held} with other files than {self.source}. " + "A published version cannot change, so other files need a new version." + ) + return message if self.hint is None else f"{message} {self.hint}" + + class TimeNetAccessError(TimeNetError): """A dataset needs credentialed or restricted access, so only the user's own registry may hold its data. diff --git a/packages/timenet/src/timenet/progress.py b/packages/timenet/src/timenet/progress.py new file mode 100644 index 000000000..1193cf660 --- /dev/null +++ b/packages/timenet/src/timenet/progress.py @@ -0,0 +1,75 @@ +"""Typed progress events for copies and builds. + +A function that copies or builds releases takes an optional :data:`ProgressCallback` and sends it +these events. It never prints. A front end, such as a CLI, decides how to show them. + +A step is the copy or the build of one release. Each step sends :class:`StepStarted` first and +:class:`StepFinished` when it is done. A step that fails sends no :class:`StepFinished`. +""" + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Literal + +from timenet.types.references import DatasetRef + + +@dataclass(frozen=True) +class StepStarted: + """A step started to copy or build one release. + + Attributes: + release: The release that the step copies or builds. + action: ``"copy"`` or ``"build"``. + """ + + release: DatasetRef + action: Literal["copy", "build"] + + +@dataclass(frozen=True) +class StepFinished: + """A step finished the copy or the build of one release. + + Attributes: + release: The release that the step copied or built. + action: ``"copy"`` or ``"build"``. + """ + + release: DatasetRef + action: Literal["copy", "build"] + + +@dataclass(frozen=True) +class BytesMoved: + """Bytes of a transfer moved. + + Attributes: + url: Where the bytes come from or go to. + done: The number of bytes moved so far. + total: The number of bytes of the whole transfer, or ``None`` if it is not known. + """ + + url: str + done: int + total: int | None + + +@dataclass(frozen=True) +class ShardWritten: + """A build wrote shards of a release. + + Attributes: + release: The release that the build writes. + count: The number of shards written so far. + """ + + release: DatasetRef + count: int + + +ProgressEvent = StepStarted | StepFinished | BytesMoved | ShardWritten +"""One progress event.""" + +ProgressCallback = Callable[[ProgressEvent], None] +"""A function that receives each progress event.""" diff --git a/packages/timenet/src/timenet/registry/copy.py b/packages/timenet/src/timenet/registry/copy.py new file mode 100644 index 000000000..b9ec678e9 --- /dev/null +++ b/packages/timenet/src/timenet/registry/copy.py @@ -0,0 +1,225 @@ +"""Copy releases and their payload sources between registries. + +A payload source is a release whose data files another release reads. +Manifests must match for existing releases, except for their build environments. +""" + +from pathlib import Path +import tempfile + +from timenet.errors import TimeNetMissingParentError, TimeNetVersionConflictError +from timenet.manifest import Manifest, same_release +from timenet.progress import ProgressCallback, StepFinished, StepStarted +from timenet.registry._paths import release_dir +from timenet.registry.base import BaseRegistry +from timenet.registry.dependencies import find_payload_sources, validate_manifest_reference +from timenet.registry.local import LocalRegistry +from timenet.registry.writable import WritableRegistry +from timenet.types.access import require_open_access +from timenet.types.references import DatasetRef + + +def copy_release( + reference: DatasetRef, + source: BaseRegistry, + dest: WritableRegistry, + *, + with_payload_sources: bool = True, + progress: ProgressCallback | None = None, +) -> None: + """Copy a release and its payload sources to another registry. + + Copy payload sources first, reuse complete releases in ``dest``, and fetch damaged local copies again. + All checks run before the first copy, so a failed check leaves ``dest`` unchanged. + + Args: + reference: The exact release to copy. + source: The registry to copy from. + dest: The registry to copy to. + with_payload_sources: Also copy the payload sources that the manifest of the release lists. + progress: Callback for ``StepStarted`` and ``StepFinished`` events for each copied or repaired release. + + Raises: + TimeNetVersionConflictError: If ``dest`` holds the release or a payload source with other files or another card. + TimeNetAccessError: If a release lacks open access where the source or destination requires it. + TimeNetDatasetNotFoundError: If ``source`` does not hold the release. + TimeNetMissingParentError: If ``source`` lacks a payload source and ``dest`` lacks a complete copy. + TimeFFormatError: If a manifest names the wrong release or pinned files differ. + """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers + manifests = _manifests(reference, source, dest, with_payload_sources=with_payload_sources) + # Check every release before anything is copied, so that a refusal leaves dest unchanged. A payload + # source that only dest holds is complete there, so it is never pending. + pending = [ref for ref, manifest in manifests.items() if _needs_copy(reference, ref, manifest, source, dest)] + _check_access(manifests, set(pending), source, dest) + if not pending: + return + # Only a copy from a non-local source to a non-local dest downloads into this directory, each + # pending release on its own. + with tempfile.TemporaryDirectory(prefix="timenet-copy-", ignore_cleanup_errors=True) as staging: + for ref in pending: + if progress is not None: + progress(StepStarted(ref, "copy")) + _copy_one(ref, manifests[ref], source, dest, Path(staging)) + if progress is not None: + progress(StepFinished(ref, "copy")) + + +def _copy_one( + reference: DatasetRef, manifest: Manifest, source: BaseRegistry, dest: WritableRegistry, staging: Path +) -> None: + """Copy one release without its payload sources to ``dest``, or repair its local copy. + + Args: + reference: The release to copy. + manifest: The release's manifest from ``source``. + source: The registry to copy from. + dest: The registry to copy to. + staging: Temporary download root when both registries are remote. + """ + # A local registry is a folder that the copy reads or writes directly. + if isinstance(dest, LocalRegistry): + source.download_release(reference, release_dir(dest.local_root, reference), manifest=manifest) + return + if isinstance(source, LocalRegistry): + directory = release_dir(source.root, reference) + else: + # put_release() needs only the release directory, and the walk checked the pins. So the payload + # sources, which dest may hold already, are not downloaded. The download checks each file by size + # and checksum. + directory = release_dir(staging, reference) + source.download_release(reference, directory, manifest=manifest) + dest.put_release(directory) + + +def _check_access( + manifests: dict[DatasetRef, Manifest], fetched: set[DatasetRef], source: BaseRegistry, dest: WritableRegistry +) -> None: + """Check release access against the source and destination requirements. + + Check all releases against ``dest`` and fetched releases against ``source``. + + Args: + manifests: Release references mapped to their manifests. + fetched: The releases that the copy fetches from ``source``. + source: The registry to copy from. + dest: The registry to copy to. + + Raises: + TimeNetAccessError: If a release or its lineage lacks open access where a registry requires it. + """ # ruff: ignore[docstring-extraneous-exception] - raised by require_open_access + for reference, manifest in manifests.items(): + dest_needs_open = dest.requires_open_access + source_needs_open = reference in fetched and source.requires_open_access + if dest_needs_open or source_needs_open: + require_open_access(manifest.metadata, manifest.lineage) + + +def _manifests( + reference: DatasetRef, source: BaseRegistry, dest: WritableRegistry, *, with_payload_sources: bool +) -> dict[DatasetRef, Manifest]: + """Collect the manifests of a release and of the payload sources that its manifest lists. + + Use a complete copy in ``dest`` when ``source`` does not serve a payload source. + + Args: + reference: The requested release. + source: The registry to copy from. + dest: The registry to copy to. + with_payload_sources: Also collect the payload sources of the release. + + Returns: + Manifests keyed by release, with the payload sources before the release. + + Raises: + TimeNetAccessError: If ``source`` lists a payload source without serving it and ``dest`` lacks a complete copy. + TimeNetMissingParentError: If neither ``source`` nor ``dest`` holds a complete payload source. + """ # ruff: ignore[docstring-extraneous-exception] - raised by require_open_access + manifest = source.get_manifest(reference.dataset_id, str(reference.version)) + validate_manifest_reference(manifest, reference) + if not with_payload_sources: + return {reference: manifest} + sources = find_payload_sources( + reference, manifest, source, find=source.find_manifest, held=lambda ref: _complete_copy(ref, dest) + ) + for missing in sources.missing: + listed = source.find_manifest(missing) + if listed is not None: + # The source lists the payload source, but does not serve its files. + require_open_access(listed.metadata, listed.lineage) + raise TimeNetMissingParentError(missing, reference, str(source), payload_source=True) + return {**sources.fetched, **sources.held, reference: manifest} + + +def _complete_copy(reference: DatasetRef, dest: WritableRegistry) -> Manifest | None: + """Return the manifest of a complete copy in a registry. + + Args: + reference: The release. + dest: The registry to look in. + + Returns: + The manifest, or ``None`` if ``dest`` lacks the release or a local file is missing or resized. + """ + held = dest.find_manifest(reference) + return held if held is not None and _complete(reference, held, dest) else None + + +def _needs_copy( + requested: DatasetRef, + reference: DatasetRef, + manifest: Manifest, + source: BaseRegistry, + dest: WritableRegistry, +) -> bool: + """Check whether a release needs copying or local repair. + + Args: + requested: The requested release, for the error message. + reference: The release to check. + manifest: The release's manifest from ``source``, or from ``dest`` if only ``dest`` holds it. + source: The source registry, for the error message. + dest: The registry to copy to. + + Returns: + ``True`` if ``dest`` lacks the release or its local copy has a missing or resized file. + + Raises: + TimeNetVersionConflictError: If ``dest`` holds the same version with other files or another card. + """ + held = dest.find_manifest(reference) + if held is None: + return True + files_differ = _file_records(held) != _file_records(manifest) + if not files_differ and same_release(held, manifest): + return not _complete(reference, manifest, dest) + needed_by = None if reference == requested else requested + raise TimeNetVersionConflictError(reference, str(dest), str(source), needed_by, "files" if files_differ else "card") + + +def _complete(reference: DatasetRef, manifest: Manifest, dest: WritableRegistry) -> bool: + """Check if a stored release is complete. + + A local copy is complete when every file has its recorded size. + Treat releases in non-local registries as complete. + + Args: + reference: The release to check. + manifest: The manifest that records the files in ``dest``. + dest: The registry that holds the release. + + Returns: + ``True`` for a complete local copy or any release in a non-local registry. + """ + return not isinstance(dest, LocalRegistry) or dest.damage(reference, manifest) is None + + +def _file_records(manifest: Manifest) -> frozenset[tuple[str, int, str]]: + """Collect each file's path, size, and checksum from a manifest. + + Args: + manifest: The manifest to read. + + Returns: + A frozen set of ``(path, size, checksum)`` entries. + """ + return frozenset((part.path, part.size, part.checksum) for part in manifest.files.all_files()) diff --git a/packages/timenet/src/timenet/registry/dependencies.py b/packages/timenet/src/timenet/registry/dependencies.py index 8836975e7..226d9c99b 100644 --- a/packages/timenet/src/timenet/registry/dependencies.py +++ b/packages/timenet/src/timenet/registry/dependencies.py @@ -9,7 +9,7 @@ """ from collections.abc import Callable, Iterable -from dataclasses import replace +from dataclasses import dataclass, field, replace from pathlib import Path from typing import TYPE_CHECKING @@ -27,6 +27,7 @@ from timenet.registry._paths import release_dir from timenet.registry.shared import complete_copy from timenet.registry.version import DatasetVersion +from timenet.types.access import Access, effective_access from timenet.types.reference_keys import SourceKey from timenet.types.references import DatasetRef @@ -143,6 +144,74 @@ def attach_payload_sources( return replace(version, payload_sources=sources) +@dataclass(frozen=True) +class CopySources: + """The payload sources of a release, by where a copy of the release finds them.""" + + fetched: dict[DatasetRef, Manifest] = field(default_factory=dict) + """The payload sources that the registry serves, with their manifests there. The copy brings them.""" + held: dict[DatasetRef, Manifest] = field(default_factory=dict) + """The other payload sources that the destination holds as complete copies, with their manifests.""" + missing: dict[DatasetRef, PayloadSource] = field(default_factory=dict) + """The payload sources that neither holds, with the pins of the release on each.""" + + +def find_payload_sources( + release: DatasetRef, + manifest: Manifest, + registry: "BaseRegistry", + *, + find: Callable[[DatasetRef], Manifest | None], + held: Callable[[DatasetRef], Manifest | None], +) -> CopySources: + """Find each payload source of a release in the registry to copy from, else in the destination. + + A release needs exactly the payload sources that its manifest lists, never their own payload sources. + + Args: + release: The release to copy. + manifest: The manifest of ``release``. + registry: The registry to copy from. + find: Return the manifest of a release in ``registry``, or ``None`` if it lacks the release. + held: Return the manifest of a complete copy in the destination, or ``None``. + + Returns: + The payload sources, by where the copy finds them. + + Raises: + TimeFFormatError: If a payload source that the copy finds differs from the files that ``release`` pins. + """ # ruff: ignore[docstring-extraneous-exception] - raised by check_pins + sources = CopySources() + for pinned in manifest.payload_sources.values(): + source = pinned.release + found = find(source) + if found is not None and _serves(registry, found): + sources.fetched[source] = found + elif (found := held(source)) is not None: + sources.held[source] = found + else: + sources.missing[source] = pinned + continue + check_pins(release, pinned, found) + return sources + + +def _serves(registry: "BaseRegistry", manifest: Manifest) -> bool: + """Check whether a registry serves the files of a release that it holds. + + A registry that requires open access serves a release only if its card and each card in its lineage are open. + + Args: + registry: The registry that holds the release. + manifest: The manifest of the release in ``registry``. + + Returns: + ``True`` if the registry serves the files. + """ + open_access = effective_access(manifest.metadata, manifest.lineage) is Access.OPEN + return open_access or not registry.requires_open_access + + def check_pins(release: DatasetRef, pinned: PayloadSource, manifest: Manifest) -> None: """Check that a payload source still records each pinned file as the release pins it. diff --git a/packages/timenet/tests/registry/test_copy.py b/packages/timenet/tests/registry/test_copy.py new file mode 100644 index 000000000..a82858fe6 --- /dev/null +++ b/packages/timenet/tests/registry/test_copy.py @@ -0,0 +1,467 @@ +"""``copy_release()`` copies a release with its payload sources and never changes a published version. + +The ``source`` registry holds a parent from ``make_dataset()`` and a child that keeps one of its +Records, so the child reads payloads that the parent stores: the parent is a payload source of the +child. The ``chain_source`` registry adds one level: a grandchild reads a middle release, which reads +a grandparent. +""" + +from pathlib import Path +import shutil +import tempfile + +from _fake_registry import build_service_fake +import numpy as np +import pytest + +from timenet.dataset import Record, Signal, Source, TimeFDataset +from timenet.errors import ( + TimeFFormatError, + TimeNetAccessError, + TimeNetMissingParentError, + TimeNetVersionConflictError, +) +from timenet.progress import ProgressEvent, StepFinished, StepStarted +from timenet.reader import TimeFReader +from timenet.registry import LocalRegistry, RemoteRegistry, WritableRegistry, copy_release, open_local_version +from timenet.registry.shared import complete_copy +from timenet.testing import make_dataset +from timenet.types import Access, Annotation, DatasetMetadata, DatasetSchema, License, TimeSeriesSpec, Version, ureg +from timenet.types.references import DatasetRef +from timenet.writer import TimeFWriter + + +parent_ref = DatasetRef.parse("timenet/hello-world@1.0.0") +child_ref = DatasetRef.parse("test/child@1.0.0") +parent = (parent_ref.dataset_id, str(parent_ref.version)) +child = (child_ref.dataset_id, str(child_ref.version)) + +grandparent_ref = DatasetRef.parse("test/grandparent@1.0.0") +middle_ref = DatasetRef.parse("test/middle@1.0.0") +grandchild_ref = DatasetRef.parse("test/grandchild@1.0.0") +chain = [(ref.dataset_id, str(ref.version)) for ref in (grandparent_ref, middle_ref, grandchild_ref)] + +_SPEC = TimeSeriesSpec(spec_type="env", name="env", unit_value=ureg.dimensionless) +_CREDENTIALED = {"access": Access.CREDENTIALED, "access_url": "https://example.org/access"} +_OPEN = {"access": Access.OPEN, "access_url": None} + + +def _derive(loaded: TimeFDataset, dataset_id: str, records, **card) -> TimeFDataset: + """Return a release under ``dataset_id`` that keeps read Records and their ids, as an edit does. + + ``card`` sets other fields of the card of the release. + """ + return TimeFDataset.from_parts( + metadata=loaded.metadata.model_copy(update={"dataset_id": dataset_id, **card}), + records=records, + tasks=(), + schema=DatasetSchema(), + last_ids=loaded.last_ids, + ) + + +def _write(root: Path, dataset: TimeFDataset) -> Path: + dataset.derive_schema() + with TimeFWriter(root, dataset) as writer: + writer.write() + return root / dataset.metadata.dataset_id / str(dataset.metadata.dataset_version) + + +def _write_parent_and_child(root: Path, *, credentialed: DatasetRef | None = None) -> None: + """Write the parent and a child that reads payloads of the parent under ``root``. + + With ``credentialed``, the card of that release declares credentialed access, and the other card + declares open access. + """ + dataset = make_dataset() + if credentialed == parent_ref: + dataset._metadata = dataset.metadata.model_copy(update=_CREDENTIALED) + parent_dir = _write(root, dataset) + with TimeFReader(open_local_version(parent_dir)) as reader: + loaded = reader.read() + card = _CREDENTIALED if credentialed == child_ref else _OPEN + _write(root, _derive(loaded, child_ref.dataset_id, [loaded.records[0]], **card)) + + +def _release(dataset_id: str, records: list[Record]) -> TimeFDataset: + dataset = TimeFDataset( + metadata=DatasetMetadata( + dataset_id=dataset_id, + dataset_version=Version(1, 0, 0), + name=dataset_id, + description="d", + license=License.MIT, + ) + ) + for record in records: + dataset.add_record(record=record) + return dataset + + +def _record(name: str) -> Record: + """Return a named Record whose Signal values are held in memory.""" + signal = Signal.from_irregular( + np.arange(3, dtype=np.float32), time_offsets_us=[0, 1_000, 2_000], spec=_SPEC, name=name + ) + return Record(sources=(Source(name="env", signals=(signal,)),), metadata={"name": name}) + + +def _write_chain(root: Path, *, grandchild_reads_grandparent: bool) -> None: + """Write a grandparent, a middle release that reads it, and a grandchild that reads the middle release. + + The middle release keeps the grandparent's Record ``"a"`` and stores its own Record ``"b"``. The + grandchild keeps ``"b"``, so it reads only the middle release. With ``grandchild_reads_grandparent``, + it also keeps ``"a"``, so it reads the grandparent too. + """ + grandparent_dir = _write(root, _release(grandparent_ref.dataset_id, [_record("a")])) + with TimeFReader(open_local_version(grandparent_dir)) as reader: + loaded = reader.read() + middle = _derive(loaded, middle_ref.dataset_id, loaded.records) + middle.add_record(record=_record("b")) + middle_dir = _write(root, middle) + with TimeFReader(open_local_version(middle_dir)) as reader: + loaded = reader.read() + kept = [record for record in loaded.records if grandchild_reads_grandparent or record.metadata["name"] == "b"] + _write(root, _derive(loaded, grandchild_ref.dataset_id, kept)) + + +def _payload_sources(registry: WritableRegistry, reference: DatasetRef) -> list[DatasetRef]: + manifest = registry.get_manifest(reference.dataset_id, str(reference.version)) + return [pinned.release for pinned in manifest.payload_sources.values()] + + +def _other_build(dataset_id: str) -> TimeFDataset: + """Return a release of ``dataset_id`` at version 1.0.0 whose files differ from the source's copy.""" + dataset = make_dataset() + dataset._metadata = dataset.metadata.model_copy(update={"dataset_id": dataset_id}) + dataset.records[0].annotate(Annotation(key="build", value="another")) + return dataset + + +def _remote(tmp_path: Path, name: str, *, token: str | None = None) -> tuple[RemoteRegistry, Path]: + """Return a remote registry over the service fake, and the directory that the fake serves.""" + root = tmp_path / name + root.mkdir(exist_ok=True) + transport, _ = build_service_fake(root, token=token) + registry = RemoteRegistry( + f"http://{name}.local", token=token, cache_dir=tmp_path / f"{name}-cache", transport=transport + ) + return registry, root + + +@pytest.fixture +def source(tmp_path) -> LocalRegistry: + root = tmp_path / "source" + _write_parent_and_child(root) + return LocalRegistry(root) + + +@pytest.fixture +def chain_source(tmp_path) -> LocalRegistry: + root = tmp_path / "chain" + _write_chain(root, grandchild_reads_grandparent=False) + registry = LocalRegistry(root) + assert _payload_sources(registry, grandchild_ref) == [middle_ref] + assert _payload_sources(registry, middle_ref) == [grandparent_ref] + return registry + + +@pytest.fixture(params=["local", "remote"]) +def dest(request, tmp_path) -> WritableRegistry: + if request.param == "local": + return LocalRegistry(tmp_path / "dest") + registry, _ = _remote(tmp_path, "dest", token="tok_rw") + return registry + + +def test_copy_release_copies_a_release_and_its_payload_sources(source, dest): + copy_release(child_ref, source, dest) + assert dest.exists(*child) and dest.exists(*parent) + assert dest.get_manifest(*child) == source.get_manifest(*child) + assert dest.get_manifest(*parent) == source.get_manifest(*parent) + + +def test_copy_release_does_nothing_for_the_same_release(source, dest, monkeypatch): + copy_release(child_ref, source, dest) + + def unexpected(*args, **kwargs): + pytest.fail("a copy of the same release must not download or upload it again") + + monkeypatch.setattr(dest, "put_release", unexpected) + monkeypatch.setattr(source, "download_release", unexpected) + events: list[ProgressEvent] = [] + copy_release(child_ref, source, dest, progress=events.append) # no error, no second upload + assert events == [] + + +def _rewrite_child_manifest(dest: LocalRegistry, **update) -> Path: + """Rewrite the manifest of the child in ``dest`` with ``update``; return the manifest file.""" + path = dest.root / child_ref.dataset_id / str(child_ref.version) / "manifest.json" + path.write_text(dest.get_manifest(*child).model_copy(update=update).to_json()) + return path + + +def test_copy_release_refuses_a_release_with_the_same_files_and_another_card(source, tmp_path): + dest = LocalRegistry(tmp_path / "dest") + copy_release(child_ref, source, dest) + card = dest.get_manifest(*child).metadata.model_copy(update={"description": "Another description"}) + path = _rewrite_child_manifest(dest, metadata=card) + before = path.read_bytes() + # A fresh registry: the rewrite can keep the mtime of the first write, so a cached manifest could stay. + with pytest.raises(TimeNetVersionConflictError, match="another card than") as raised: + copy_release(child_ref, source, LocalRegistry(dest.root)) + assert raised.value.difference == "card" + assert path.read_bytes() == before + + +def test_copy_release_ignores_the_build_environment(source, tmp_path): + dest = LocalRegistry(tmp_path / "dest") + copy_release(child_ref, source, dest) + _rewrite_child_manifest(dest, build_env={"python": "0.0.0"}) + events: list[ProgressEvent] = [] + copy_release(child_ref, source, LocalRegistry(dest.root), progress=events.append) + assert events == [] + + +def test_copy_release_refuses_another_build_of_the_version(source, dest): + # the destination holds child_ref built from other records + dest.put_dataset(_other_build(child_ref.dataset_id)) + before = dest.get_manifest(*child) + with pytest.raises(TimeNetVersionConflictError, match="new version") as raised: + copy_release(child_ref, source, dest) + assert (raised.value.release, raised.value.needed_by) == (child_ref, None) + assert (raised.value.registry, raised.value.source) == (str(dest), str(source)) + assert raised.value.difference == "files" # the card differs too, but the files come first + assert dest.get_manifest(*child) == before + assert not dest.exists(*parent) # nothing is copied, not even the payload source + + +def test_copy_release_refuses_a_payload_source_with_other_files(source, tmp_path): + dest = LocalRegistry(tmp_path / "dest") + # the destination holds parent_ref with other files + dest.put_dataset(_other_build(parent_ref.dataset_id)) + before = dest.get_manifest(*parent) + with pytest.raises(TimeNetVersionConflictError, match=str(parent_ref)) as raised: + copy_release(child_ref, source, dest) + assert (raised.value.release, raised.value.needed_by) == (parent_ref, child_ref) + assert not dest.exists(*child) + assert dest.get_manifest(*parent) == before + + +def _without_parent(source: LocalRegistry) -> LocalRegistry: + """Remove the parent from ``source``, and return a new registry over its directory. + + The new registry does not keep a cached manifest of the parent. + """ + shutil.rmtree(source.root / parent_ref.dataset_id) + return LocalRegistry(source.root) + + +def test_copy_release_reads_a_payload_source_that_only_dest_holds(source, dest): + copy_release(parent_ref, source, dest) + before = dest.get_manifest(*parent) + events: list[ProgressEvent] = [] + copy_release(child_ref, _without_parent(source), dest, progress=events.append) + assert dest.get_manifest(*child) == source.get_manifest(*child) + assert dest.get_manifest(*parent) == before + assert events == [StepStarted(child_ref, "copy"), StepFinished(child_ref, "copy")] + + +def test_copy_release_refuses_a_payload_source_that_only_dest_holds_with_other_files(source, dest): + # the destination holds parent_ref with other payload files than the child pins + dest.put_dataset(_release(parent_ref.dataset_id, [_record("other")])) + before = dest.get_manifest(*parent) + with pytest.raises(TimeFFormatError, match=f"payload source {parent_ref} differs"): + copy_release(child_ref, _without_parent(source), dest) + assert not dest.exists(*child) + assert dest.get_manifest(*parent) == before + + +def test_copy_release_checks_the_access_of_a_payload_source_that_only_dest_holds(tmp_path): + root = tmp_path / "source" + _write_parent_and_child(root, credentialed=parent_ref) + dest, dest_root = _remote(tmp_path, "dest", token="tok_rw") + # the service holds the parent, whose card declares credentialed access + shutil.copytree(root / parent_ref.dataset_id, dest_root / parent_ref.dataset_id) + with pytest.raises(TimeNetAccessError) as raised: + copy_release(child_ref, _without_parent(LocalRegistry(root)), dest) + assert raised.value.dataset == parent_ref + assert not dest.exists(*child) + + +def test_copy_release_refuses_a_payload_source_that_no_registry_holds(source, dest): + with pytest.raises(TimeNetMissingParentError) as raised: + copy_release(child_ref, _without_parent(source), dest) + assert (raised.value.missing, raised.value.needed_by) == (parent_ref, child_ref) + assert not dest.exists(*child) + + +def test_copy_release_refuses_a_payload_source_that_only_dest_holds_damaged(source, tmp_path): + dest = LocalRegistry(tmp_path / "dest") + copy_release(parent_ref, source, dest) + parent_dir = dest.root / parent_ref.dataset_id / str(parent_ref.version) + (parent_dir / dest.get_manifest(*parent).files.control[0].path).unlink() + with pytest.raises(TimeNetMissingParentError) as raised: + copy_release(child_ref, _without_parent(source), dest) + assert (raised.value.missing, raised.value.needed_by) == (parent_ref, child_ref) + assert not dest.exists(*child) + + +def test_copy_release_repairs_a_damaged_local_copy(source, tmp_path): + dest = LocalRegistry(tmp_path / "dest") + copy_release(child_ref, source, dest) + child_dir = dest.root / child_ref.dataset_id / str(child_ref.version) + manifest = dest.get_manifest(*child) + (child_dir / manifest.files.control[0].path).unlink() + assert not complete_copy(child_dir, manifest) + events: list[ProgressEvent] = [] + copy_release(child_ref, source, dest, progress=events.append) + assert complete_copy(child_dir, manifest) + assert events == [StepStarted(child_ref, "copy"), StepFinished(child_ref, "copy")] + + +def test_copy_release_reports_each_copied_release(source, tmp_path): + events: list[ProgressEvent] = [] + copy_release(child_ref, source, LocalRegistry(tmp_path / "dest"), progress=events.append) + assert [(e.action, e.release) for e in events if isinstance(e, StepStarted)] == [ + ("copy", parent_ref), + ("copy", child_ref), + ] + assert events == [ + StepStarted(parent_ref, "copy"), + StepFinished(parent_ref, "copy"), + StepStarted(child_ref, "copy"), + StepFinished(child_ref, "copy"), + ] + + +def test_copy_release_copies_only_the_payload_sources_that_the_release_lists(chain_source, dest): + # The grandchild reads the middle release only. The source lacks the grandparent, which the middle reads. + shutil.rmtree(chain_source.root / grandparent_ref.dataset_id) + events: list[ProgressEvent] = [] + copy_release(grandchild_ref, LocalRegistry(chain_source.root), dest, progress=events.append) + assert [e.release for e in events if isinstance(e, StepStarted)] == [middle_ref, grandchild_ref] + assert not dest.exists(*chain[0]) + + +def test_copy_release_without_payload_sources_copies_the_release_alone(chain_source, dest): + copy_release(middle_ref, chain_source, dest, with_payload_sources=False) + assert dest.get_manifest(*chain[1]) == chain_source.get_manifest(*chain[1]) + assert not dest.exists(*chain[0]) + + +def test_copy_release_copies_a_shared_payload_source_once(tmp_path): + root = tmp_path / "chain" + _write_chain(root, grandchild_reads_grandparent=True) + shared_source = LocalRegistry(root) + assert set(_payload_sources(shared_source, grandchild_ref)) == {grandparent_ref, middle_ref} + events: list[ProgressEvent] = [] + copy_release(grandchild_ref, shared_source, LocalRegistry(tmp_path / "dest"), progress=events.append) + assert [e.release for e in events if isinstance(e, StepStarted)] == [grandparent_ref, middle_ref, grandchild_ref] + + +def test_copy_release_between_remote_registries_removes_its_download(tmp_path, monkeypatch): + remote_source, source_root = _remote(tmp_path, "source") + _write_parent_and_child(source_root) + dest, _ = _remote(tmp_path, "dest", token="tok_rw") + downloads = tmp_path / "tmp" + downloads.mkdir() + monkeypatch.setattr(tempfile, "tempdir", str(downloads)) + copy_release(child_ref, remote_source, dest) + assert dest.get_manifest(*child) == remote_source.get_manifest(*child) + assert dest.get_manifest(*parent) == remote_source.get_manifest(*parent) + assert list(downloads.iterdir()) == [] + + +def test_copy_release_between_remote_registries_downloads_only_the_releases_that_dest_lacks(tmp_path, monkeypatch): + remote_source, source_root = _remote(tmp_path, "source") + _write_parent_and_child(source_root) + dest, _ = _remote(tmp_path, "dest", token="tok_rw") + copy_release(parent_ref, remote_source, dest) + downloads: list[DatasetRef] = [] + download = remote_source.download_release + + def counting(reference, *args, **kwargs): + downloads.append(reference) + return download(reference, *args, **kwargs) + + monkeypatch.setattr(remote_source, "download_release", counting) + copy_release(child_ref, remote_source, dest) + assert downloads == [child_ref] # dest already holds the parent + assert dest.get_manifest(*child) == remote_source.get_manifest(*child) + + +def test_copy_release_removes_its_download_after_an_error(tmp_path, monkeypatch): + remote_source, source_root = _remote(tmp_path, "source") + _write_parent_and_child(source_root) + dest, _ = _remote(tmp_path, "dest", token="tok_rw") + downloads = tmp_path / "tmp" + downloads.mkdir() + monkeypatch.setattr(tempfile, "tempdir", str(downloads)) + + def refuse(directory): + raise RuntimeError("upload failed") + + monkeypatch.setattr(dest, "put_release", refuse) + with pytest.raises(RuntimeError, match="upload failed"): + copy_release(child_ref, remote_source, dest) + assert list(downloads.iterdir()) == [] + + +@pytest.mark.parametrize("remote", [False, True], ids=["local", "remote"]) +def test_a_forced_download_keeps_the_old_copy_if_the_final_rename_fails(remote, tmp_path, monkeypatch): + if remote: + registry, root = _remote(tmp_path, "source") + else: + root = tmp_path / "source" + registry = LocalRegistry(root) + _write_parent_and_child(root) + manifest = registry.get_manifest(*parent) + directory = tmp_path / "copy" / parent_ref.dataset_id / str(parent_ref.version) + registry.download_release(parent_ref, directory, manifest=manifest) + rename = Path.replace + + def replace(self, target): + if ".tmp-" in self.name: + raise OSError("rename failed") + return rename(self, target) + + monkeypatch.setattr(Path, "replace", replace) + with pytest.raises(OSError, match="rename failed"): + registry.download_release(parent_ref, directory, force=True, manifest=manifest) + assert complete_copy(directory, manifest) + + +@pytest.mark.parametrize("credentialed", [child_ref, parent_ref], ids=["release", "payload-source"]) +def test_copy_release_refuses_a_release_that_is_not_open_for_a_registry_that_is_not_local(credentialed, tmp_path): + root = tmp_path / "source" + _write_parent_and_child(root, credentialed=credentialed) + dest, _ = _remote(tmp_path, "dest", token="tok_rw") + with pytest.raises(TimeNetAccessError) as raised: + copy_release(child_ref, LocalRegistry(root), dest) + assert raised.value.dataset == credentialed + assert not dest.exists(*child) + assert not dest.exists(*parent) + + +@pytest.mark.parametrize("credentialed", [child_ref, parent_ref], ids=["release", "payload-source"]) +def test_copy_release_copies_a_release_that_is_not_open_between_local_registries(credentialed, tmp_path): + root = tmp_path / "source" + _write_parent_and_child(root, credentialed=credentialed) + dest = LocalRegistry(tmp_path / "dest") + copy_release(child_ref, LocalRegistry(root), dest) + assert dest.get_manifest(*child).metadata.access is ( + Access.CREDENTIALED if credentialed == child_ref else Access.OPEN + ) + assert dest.exists(*parent) + + +def test_copy_release_refuses_to_fetch_a_release_that_is_not_open_from_a_registry_that_is_not_local(tmp_path): + remote_source, source_root = _remote(tmp_path, "source") + _write_parent_and_child(source_root, credentialed=child_ref) + dest = LocalRegistry(tmp_path / "dest") + with pytest.raises(TimeNetAccessError) as raised: + copy_release(child_ref, remote_source, dest) + assert raised.value.dataset == child_ref + assert not dest.exists(*child) + assert not dest.exists(*parent) # the open parent is not copied either: every check runs first diff --git a/packages/timenet/tests/test_errors.py b/packages/timenet/tests/test_errors.py index 1df76869e..e895315e1 100644 --- a/packages/timenet/tests/test_errors.py +++ b/packages/timenet/tests/test_errors.py @@ -15,6 +15,7 @@ TimeNetFileError, TimeNetMissingParentError, TimeNetRegistryError, + TimeNetVersionConflictError, ) from timenet.types import Access from timenet.types.access import AccessRequirement @@ -113,6 +114,18 @@ def test_catchable_as_base(): {"requirements": ()}, "org/a@1.0.0 is credentialed.", ), + ( + TimeNetVersionConflictError(_A, "registry here", "registry there"), + {"release": _A, "registry": "registry here", "source": "registry there", "needed_by": None}, + "registry here already holds org/a@1.0.0 with other files than registry there. " + "A published version cannot change, so other files need a new version.", + ), + ( + TimeNetVersionConflictError(_A, "registry here", "registry there", _B), + {"release": _A, "needed_by": _B}, + "registry here already holds org/a@1.0.0, a payload source of org/b@2.0.0, with other files than " + "registry there. A published version cannot change, so other files need a new version.", + ), ], ) def test_an_error_with_facts_ends_with_the_hint_of_its_front_end(error, facts, message): diff --git a/packages/timenet/tests/test_progress.py b/packages/timenet/tests/test_progress.py new file mode 100644 index 000000000..2b89ae30d --- /dev/null +++ b/packages/timenet/tests/test_progress.py @@ -0,0 +1,12 @@ +from dataclasses import FrozenInstanceError + +import pytest + +from timenet.progress import StepStarted +from timenet.types.references import DatasetRef + + +def test_progress_events_are_frozen(): + event = StepStarted(DatasetRef.parse("a/b@1.0.0"), "copy") + with pytest.raises(FrozenInstanceError): + event.action = "build" # ty: ignore[invalid-assignment] diff --git a/zensical.toml b/zensical.toml index 283c61573..a10b4762d 100644 --- a/zensical.toml +++ b/zensical.toml @@ -22,7 +22,7 @@ nav = [ { "Datasets" = [{ "Use datasets" = [{ "The client (SDK)" = "client.md" }, { "Registries" = "registry.md" }, { "Load into your stack" = "usage.md" }] }, { "Build datasets" = [{ "Connectors" = "connectors.md" }, { "Build & publish" = "build.md" }] }, { "Catalog" = [{ "Datasets" = "catalog/datasets.md" }, { "Benchmarks" = "catalog/benchmarks.md" }, { "Licensing" = "catalog/licensing.md" }] }] }, { "CLI" = [{ "Overview" = "cli/index.md" }, { "timenet (consumer)" = "cli/timenet.md" }, { "timenet-build (producer)" = "cli/build.md" }] }, # >>> api-nav: generated by scripts/gen_api_docs.py — do not edit >>> - { "API Reference" = [{ "Overview" = "api/index.md" }, { "timenet.builders" = "api/builders.md" }, { "timenet.client" = "api/client.md" }, { "timenet.config" = "api/config.md" }, { "timenet.connectors" = "api/connectors.md" }, { "timenet.dataset" = "api/dataset.md" }, { "timenet.engine" = "api/engine.md" }, { "timenet.errors" = "api/errors.md" }, { "timenet.format" = "api/format.md" }, { "timenet.json" = "api/json.md" }, { "timenet.manifest" = "api/manifest.md" }, { "timenet.provenance" = "api/provenance.md" }, { "timenet.reader" = "api/reader.md" }, { "timenet.registry" = "api/registry.md" }, { "timenet.storage" = "api/storage.md" }, { "timenet.types" = "api/types.md" }, { "timenet.values_backends" = "api/values_backends.md" }, { "timenet.writer" = "api/writer.md" }, { "timenet_connectors.api" = "api/api.md" }, { "timenet_connectors.bases" = "api/bases.md" }, { "timenet_connectors.discovery" = "api/discovery.md" }, { "timenet_connectors.download" = "api/download.md" }, { "timenet_connectors.sources" = "api/sources.md" }, { "timenet_connectors.splitting" = "api/splitting.md" }, { "timenet_connectors.time_axes" = "api/time_axes.md" }] }, + { "API Reference" = [{ "Overview" = "api/index.md" }, { "timenet.builders" = "api/builders.md" }, { "timenet.client" = "api/client.md" }, { "timenet.config" = "api/config.md" }, { "timenet.connectors" = "api/connectors.md" }, { "timenet.dataset" = "api/dataset.md" }, { "timenet.engine" = "api/engine.md" }, { "timenet.errors" = "api/errors.md" }, { "timenet.format" = "api/format.md" }, { "timenet.json" = "api/json.md" }, { "timenet.manifest" = "api/manifest.md" }, { "timenet.progress" = "api/progress.md" }, { "timenet.provenance" = "api/provenance.md" }, { "timenet.reader" = "api/reader.md" }, { "timenet.registry" = "api/registry.md" }, { "timenet.storage" = "api/storage.md" }, { "timenet.types" = "api/types.md" }, { "timenet.values_backends" = "api/values_backends.md" }, { "timenet.writer" = "api/writer.md" }, { "timenet_connectors.api" = "api/api.md" }, { "timenet_connectors.bases" = "api/bases.md" }, { "timenet_connectors.discovery" = "api/discovery.md" }, { "timenet_connectors.download" = "api/download.md" }, { "timenet_connectors.sources" = "api/sources.md" }, { "timenet_connectors.splitting" = "api/splitting.md" }, { "timenet_connectors.time_axes" = "api/time_axes.md" }] }, # <<< api-nav <<< ] From 9a26329c8a01107355e091738f0ef0315cad0990 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Wed, 4 Nov 2026 12:11:09 +0100 Subject: [PATCH 2/5] refactor(registry): export copy_release() and rewrite its docstring for users copy_release() is now part of the public names of timenet.registry, next to LocalRegistry. Callers import it with `from timenet.registry import copy_release`. Its module is now the private timenet/registry/_copy.py, like _paths.py. The module docstring describes the copy in plain words and no longer lists the internal steps. The docstrings of this change say "non-local" for a registry that is not local, and the TimeNetVersionConflictError text says that a registry "already holds" the release. Part of TNET-157. --- packages/timenet/src/timenet/errors.py | 2 +- packages/timenet/src/timenet/registry/__init__.py | 2 ++ packages/timenet/src/timenet/registry/{copy.py => _copy.py} | 0 3 files changed, 3 insertions(+), 1 deletion(-) rename packages/timenet/src/timenet/registry/{copy.py => _copy.py} (100%) diff --git a/packages/timenet/src/timenet/errors.py b/packages/timenet/src/timenet/errors.py index c0a198176..4a7792dbd 100644 --- a/packages/timenet/src/timenet/errors.py +++ b/packages/timenet/src/timenet/errors.py @@ -59,7 +59,7 @@ def __str__(self) -> str: class TimeNetVersionConflictError(TimeNetError): - """A registry holds a release with other files than the copy that would go there. + """A registry already holds a release with other files than the copy that would go there. A published version cannot change. Other files need a new version. diff --git a/packages/timenet/src/timenet/registry/__init__.py b/packages/timenet/src/timenet/registry/__init__.py index eaa0a5dfc..bc13fea69 100644 --- a/packages/timenet/src/timenet/registry/__init__.py +++ b/packages/timenet/src/timenet/registry/__init__.py @@ -1,5 +1,6 @@ """Dataset registries: serve compiled TimeF versions and search over metadata.""" +from timenet.registry._copy import copy_release from timenet.registry.base import BaseRegistry from timenet.registry.dependencies import open_local_version from timenet.registry.factory import ( @@ -24,6 +25,7 @@ "RemoteRegistry", "S3Registry", "WritableRegistry", + "copy_release", "default_registry_path", "local_registry_path", "open_local_version", diff --git a/packages/timenet/src/timenet/registry/copy.py b/packages/timenet/src/timenet/registry/_copy.py similarity index 100% rename from packages/timenet/src/timenet/registry/copy.py rename to packages/timenet/src/timenet/registry/_copy.py From d3aafe385acae9f992a83937d1953b15b1778ea9 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Wed, 4 Nov 2026 18:14:27 +0100 Subject: [PATCH 3/5] refactor(registry): name the access rule of a registry, and shorten the copy docstrings BaseRegistry.requires_open_access tells whether a registry holds and serves only releases with open access. The access checks use it instead of a test for LocalRegistry. BaseRegistry.download_release() is public. TimeNetMissingParentError takes payload_source as a keyword argument. timenet.progress lists only the events that a copy sends. The docstrings of copy_release() are shorter. Part of TNET-157. --- packages/timenet/src/timenet/client.py | 6 +-- .../timenet/src/timenet/coordinator/plan.py | 14 +++---- packages/timenet/src/timenet/engine/engine.py | 4 +- packages/timenet/src/timenet/errors.py | 12 ++++-- packages/timenet/src/timenet/progress.py | 38 ++----------------- packages/timenet/src/timenet/registry/base.py | 33 ++++++++-------- .../src/timenet/registry/dependencies.py | 4 +- .../timenet/src/timenet/registry/local.py | 9 +++++ .../src/timenet/registry/remote/registry.py | 18 ++++----- packages/timenet/tests/test_client.py | 2 + packages/timenet/tests/test_engine.py | 4 ++ packages/timenet/tests/test_errors.py | 4 +- 12 files changed, 71 insertions(+), 77 deletions(-) diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index bc5ca76b5..1c20057d8 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -319,9 +319,9 @@ def _reject_unhosted_access(self, dataset_id: str, version: str | None) -> None: version: The resolved version, or ``None`` for the latest. Raises: - TimeNetAccessError: If the dataset is not open and the registry is not a local one. + TimeNetAccessError: If the dataset is not open and the registry requires open access. """ - if isinstance(self._registry, LocalRegistry): + if not self._registry.requires_open_access: return try: manifest = self._registry.get_manifest(dataset_id, version) @@ -415,7 +415,7 @@ def _prepare( manifest = _obtain(registry, dataset_id, version, miss, auto_build=auto_build, fetch_from=fetch_from) reference = DatasetRef(dataset_id=dataset_id, version=manifest.metadata.dataset_version) # A registry that is not local serves only releases with access declared ``open``. - check = None if fetch_from is None or isinstance(fetch_from, LocalRegistry) else _require_open + check = _require_open if fetch_from is not None and fetch_from.requires_open_access else None built: set[DatasetRef] = set() while True: try: diff --git a/packages/timenet/src/timenet/coordinator/plan.py b/packages/timenet/src/timenet/coordinator/plan.py index 2b38af2ab..f1ed9d4b3 100644 --- a/packages/timenet/src/timenet/coordinator/plan.py +++ b/packages/timenet/src/timenet/coordinator/plan.py @@ -259,7 +259,7 @@ def plan_parents( # If the output already holds the requested release, a reader needs only its payload sources to # open it. Its declared parents matter only to a rebuild. held = None if force_build else _manifest(output, root) - if held is None and not isinstance(output, LocalRegistry): + if held is None and output.requires_open_access: require_open_access(card) searched = str(output) if fetch_from is None else f"{output} or {fetch_from}" graph = _Graph(output, fetch_from, build_parents=build_parents, searched=searched) @@ -375,14 +375,14 @@ def _locate( except TimeNetFileError as error: manifest, damage = None, error if manifest is not None: - if not isinstance(output, LocalRegistry): + if output.requires_open_access: require_open_access(manifest.metadata, manifest.lineage) return manifest, None fetched = None if fetch_from is None else _manifest(fetch_from, reference) if fetched is not None: - # A registry that is not local holds only releases whose cards declare access: open. So the - # plan refuses another release before any download. - if not isinstance(fetch_from, LocalRegistry): + # A registry that requires open access serves only releases whose cards declare access: open. + # So the plan refuses another release before any download. + if fetch_from is not None and fetch_from.requires_open_access: require_open_access(fetched.metadata, fetched.lineage) return fetched, "fetch" if damage is not None: @@ -409,9 +409,9 @@ def _card_to_build( # ruff: ignore[too-many-arguments] TimeNetAccessError: If the output is not local and the card does not declare ``access: open``. """ # ruff: ignore[docstring-extraneous-exception] - the access error is raised by require_open_access if not build_parents: - raise TimeNetMissingParentError(reference, needed_by, pinned, searched) + raise TimeNetMissingParentError(reference, needed_by, searched, payload_source=pinned) card = _exact_card(reference) - if not isinstance(output, LocalRegistry): + if output.requires_open_access: require_open_access(card) return card diff --git a/packages/timenet/src/timenet/engine/engine.py b/packages/timenet/src/timenet/engine/engine.py index d592bf0d4..e46264409 100644 --- a/packages/timenet/src/timenet/engine/engine.py +++ b/packages/timenet/src/timenet/engine/engine.py @@ -140,7 +140,7 @@ def publish_pipeline( # ruff: ignore[too-many-arguments] version = str(metadata.dataset_version) if not force and registry.exists(dataset_id, version): return version - if not isinstance(registry, LocalRegistry): + if registry.requires_open_access: require_open_access(metadata) _check_composes(connector, metadata) @@ -149,7 +149,7 @@ def publish_pipeline( # ruff: ignore[too-many-arguments] with open_parents(metadata, registry) as context: dataset = _compose(connector, metadata, cache, context) - if not isinstance(registry, LocalRegistry): + if registry.requires_open_access: require_open_access(dataset.metadata, dataset.lineage.values()) dataset.derive_schema() registry.put_dataset( diff --git a/packages/timenet/src/timenet/errors.py b/packages/timenet/src/timenet/errors.py index 4a7792dbd..427896e24 100644 --- a/packages/timenet/src/timenet/errors.py +++ b/packages/timenet/src/timenet/errors.py @@ -12,6 +12,7 @@ sentence. """ +from functools import partial from typing import TYPE_CHECKING @@ -44,14 +45,19 @@ class TimeNetMissingParentError(TimeNetDatasetNotFoundError): hint: How the caller can obtain the release, added by the front end, or ``None``. """ - def __init__(self, missing: "DatasetRef", needed_by: "DatasetRef", payload_source: bool, searched: str) -> None: - super().__init__(missing, needed_by, payload_source, searched) + def __init__(self, missing: "DatasetRef", needed_by: "DatasetRef", searched: str, *, payload_source: bool) -> None: + super().__init__(missing, needed_by, searched, payload_source) self.missing = missing self.needed_by = needed_by self.payload_source = payload_source self.searched = searched self.hint: str | None = None + def __reduce__(self) -> tuple[object, ...]: + # Pickle passes the arguments by position, so the keyword-only flag travels in a partial. + rebuild = partial(type(self), payload_source=self.payload_source) + return rebuild, (self.missing, self.needed_by, self.searched), self.__dict__ + def __str__(self) -> str: role = "payload source" if self.payload_source else "parent release" message = f"{role} {self.missing} of {self.needed_by} is not in {self.searched}." @@ -61,7 +67,7 @@ def __str__(self) -> str: class TimeNetVersionConflictError(TimeNetError): """A registry already holds a release with other files than the copy that would go there. - A published version cannot change. Other files need a new version. + A published version cannot change, so other files need a new version. Attributes: release: The release that the registry holds with other files. diff --git a/packages/timenet/src/timenet/progress.py b/packages/timenet/src/timenet/progress.py index 1193cf660..eb1659ef6 100644 --- a/packages/timenet/src/timenet/progress.py +++ b/packages/timenet/src/timenet/progress.py @@ -1,10 +1,8 @@ """Typed progress events for copies and builds. -A function that copies or builds releases takes an optional :data:`ProgressCallback` and sends it -these events. It never prints. A front end, such as a CLI, decides how to show them. - -A step is the copy or the build of one release. Each step sends :class:`StepStarted` first and -:class:`StepFinished` when it is done. A step that fails sends no :class:`StepFinished`. +A function that copies or builds releases sends these events to an optional +:data:`ProgressCallback` and prints nothing. Each step, the copy or the build of one release, sends +:class:`StepStarted` first and :class:`StepFinished` only if it succeeds. """ from collections.abc import Callable @@ -40,35 +38,7 @@ class StepFinished: action: Literal["copy", "build"] -@dataclass(frozen=True) -class BytesMoved: - """Bytes of a transfer moved. - - Attributes: - url: Where the bytes come from or go to. - done: The number of bytes moved so far. - total: The number of bytes of the whole transfer, or ``None`` if it is not known. - """ - - url: str - done: int - total: int | None - - -@dataclass(frozen=True) -class ShardWritten: - """A build wrote shards of a release. - - Attributes: - release: The release that the build writes. - count: The number of shards written so far. - """ - - release: DatasetRef - count: int - - -ProgressEvent = StepStarted | StepFinished | BytesMoved | ShardWritten +ProgressEvent = StepStarted | StepFinished """One progress event.""" ProgressCallback = Callable[[ProgressEvent], None] diff --git a/packages/timenet/src/timenet/registry/base.py b/packages/timenet/src/timenet/registry/base.py index 18f953ad8..012d52628 100644 --- a/packages/timenet/src/timenet/registry/base.py +++ b/packages/timenet/src/timenet/registry/base.py @@ -42,6 +42,11 @@ def __str__(self) -> str: def local_root(self) -> Path: """The local registry or cache directory that holds prepared releases.""" + @property + def requires_open_access(self) -> bool: + """Whether this registry holds and serves only releases with open access.""" + return True + @abstractmethod def list_datasets(self) -> list[DatasetMetadata]: """Return the metadata of every dataset, using the latest version. The result is sorted by dataset id. @@ -143,38 +148,36 @@ def download_version( # ruff: ignore[too-many-arguments] progress_cb=progress_cb, ) - def _download_release( # ruff: ignore[too-many-arguments] + def download_release( # ruff: ignore[too-many-arguments] self, dataset_id: str, version: str, - dest_dir: str | Path, + directory: str | Path, *, force: bool = False, manifest: Manifest | None = None, progress_cb: ProgressCallback | None = None, ) -> None: - """Download a version's files into ``dest_dir``, swapping the directory in atomically. + """Download the files of one release into a local directory. - Fetches every file through :meth:`open_file` into a sibling ``.tmp-*`` staging - directory, with ``manifest.json`` last as the commit marker. It then renames the staging - directory over ``dest_dir`` in one step. An interrupted download never leaves a half-written - copy in place. Subclasses can override this method with a faster path, such as concurrent - streaming from a remote service. + The download replaces ``directory`` only after every file arrived and matched its manifest + record, so a failed download leaves no partial copy. Args: - dataset_id: The dataset id. + dataset_id: The dataset ID. version: The version string. - dest_dir: The target ``<...>//`` directory. - force: Re-download even if a copy already exists. - manifest: The parsed manifest, passed to avoid re-fetching it. Fetched if ``None``. - progress_cb: Called with each staged file's byte count, for a progress display. + directory: The directory of the release, ``//``. + force: Download again, even if ``directory`` holds a complete copy. + manifest: The manifest of the release, or ``None`` to read it from this registry. + progress_cb: Receives the size of each downloaded file, or ``None``. Raises: - TimeNetFileError: If a downloaded file's size or checksum differs from the manifest. + TimeNetFileError: If a downloaded file has another size or checksum than its manifest + records. """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_staged_file if manifest is None: manifest = self.get_manifest(dataset_id, version) - target = Path(dest_dir) + target = Path(directory) if complete_copy(target, manifest) and not force: return # already downloaded and current parent = target.parent diff --git a/packages/timenet/src/timenet/registry/dependencies.py b/packages/timenet/src/timenet/registry/dependencies.py index 226d9c99b..d4c46cc17 100644 --- a/packages/timenet/src/timenet/registry/dependencies.py +++ b/packages/timenet/src/timenet/registry/dependencies.py @@ -136,7 +136,7 @@ def attach_payload_sources( try: source = open_source(pinned.release) except TimeNetDatasetNotFoundError as error: - raise TimeNetMissingParentError(pinned.release, release, True, searched) from error + raise TimeNetMissingParentError(pinned.release, release, searched, payload_source=True) from error validate_manifest_reference(source.manifest, pinned.release) check_pins(release, pinned, source.manifest) check_file_sizes(source, pinned.files, owner=f"payload source {pinned.release} of {release}") @@ -441,7 +441,7 @@ def _download_dataset_version( # ruff: ignore[too-many-arguments] validate_manifest_reference(manifest, reference) if before_fetch is not None: before_fetch(manifest) - registry._download_release( + registry.download_release( reference.dataset_id, str(reference.version), directory, diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py index 81b3a2eaf..54979ab29 100644 --- a/packages/timenet/src/timenet/registry/local.py +++ b/packages/timenet/src/timenet/registry/local.py @@ -48,6 +48,15 @@ def local_root(self) -> Path: """The directory that holds complete local releases.""" return self._root + @property + def requires_open_access(self) -> bool: + """Whether this registry holds and serves only releases with open access. + + Returns: + ``False``: a local registry holds releases of any access. + """ + return False + def list_datasets(self) -> list[DatasetMetadata]: """Return the latest-version metadata of every dataset, sorted by id. diff --git a/packages/timenet/src/timenet/registry/remote/registry.py b/packages/timenet/src/timenet/registry/remote/registry.py index 35a2740a4..5cbc5de96 100644 --- a/packages/timenet/src/timenet/registry/remote/registry.py +++ b/packages/timenet/src/timenet/registry/remote/registry.py @@ -130,30 +130,30 @@ def _open_remote_version(self, reference: DatasetRef) -> DatasetVersion: filesystem = RegistryFileSystem(self._base_url, files, token=self._http.token, transport=self._http.transport) return DatasetVersion(manifest, pafs.PyFileSystem(pafs.FSSpecHandler(filesystem)), root) - def _download_release( # ruff: ignore[too-many-arguments] + def download_release( # ruff: ignore[too-many-arguments] self, dataset_id: str, version: str, - dest_dir: str | Path, + directory: str | Path, *, force: bool = False, manifest: Manifest | None = None, progress_cb: ProgressCallback | None = None, ) -> None: - """Download a version's files into ``dest_dir`` (used by ``TimeNet.download`` for remotes). + """Download the files of one release into a local directory. Args: - dataset_id: The ``org/name`` id. + dataset_id: The dataset ID, as ``org/name``. version: The version string. - dest_dir: The target ``<...>//`` directory. - force: Re-download even if a copy already exists. - manifest: The already-parsed manifest, passed to avoid re-fetching it; fetched if ``None``. - progress_cb: Called with each chunk's byte count as it is written, for a progress display. + directory: The directory of the release, ``//``. + force: Download again, even if ``directory`` holds a complete copy. + manifest: The manifest of the release, or ``None`` to read it from this registry. + progress_cb: Receives the size of each downloaded chunk, or ``None``. """ if manifest is None: manifest = self.get_manifest(dataset_id, version) download_version_files( - self._http, manifest, dataset_id, version, Path(dest_dir), force=force, progress_cb=progress_cb + self._http, manifest, dataset_id, version, Path(directory), force=force, progress_cb=progress_cb ) def put_release(self, from_directory: Path) -> DatasetRef: diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py index 3ea6740b2..0ad9fc02f 100644 --- a/packages/timenet/tests/test_client.py +++ b/packages/timenet/tests/test_client.py @@ -636,6 +636,8 @@ def _hosted(client, metadata, lineage=()): """Serve one manifest from a registry that is not local, so the build-your-own gate applies.""" class _Hosted: + requires_open_access = True + def get_manifest(self, dataset_id, version=None): return SimpleNamespace(metadata=metadata, lineage=lineage) diff --git a/packages/timenet/tests/test_engine.py b/packages/timenet/tests/test_engine.py index 2fb1f96bd..9efecaa79 100644 --- a/packages/timenet/tests/test_engine.py +++ b/packages/timenet/tests/test_engine.py @@ -106,6 +106,8 @@ class _ZarrConnector(_DemoConnector): def test_publish_pipeline_passes_the_requested_values_backend_to_the_registry(tmp_path): class _Registry: + requires_open_access = True + def __init__(self) -> None: self.values_backend: str | None = None @@ -484,6 +486,8 @@ def test_publish_pipeline_refuses_a_card_that_is_not_open_before_downloading(tmp class _HostedRegistry: """A registry that is not local. It serves the parents from a local registry and refuses writes.""" + requires_open_access = True + def __init__(self, root: Path) -> None: self.local = LocalRegistry(root) diff --git a/packages/timenet/tests/test_errors.py b/packages/timenet/tests/test_errors.py index e895315e1..921bc5d6c 100644 --- a/packages/timenet/tests/test_errors.py +++ b/packages/timenet/tests/test_errors.py @@ -67,12 +67,12 @@ def test_catchable_as_base(): ("error", "facts", "message"), [ ( - TimeNetMissingParentError(_A, _B, True, "here"), + TimeNetMissingParentError(_A, _B, "here", payload_source=True), {"missing": _A, "needed_by": _B, "payload_source": True}, "payload source org/a@1.0.0 of org/b@2.0.0 is not in here.", ), ( - TimeNetMissingParentError(_A, _B, False, "here"), + TimeNetMissingParentError(_A, _B, "here", payload_source=False), {"payload_source": False}, "parent release org/a@1.0.0 of org/b@2.0.0 is not in here.", ), From 4fb70cefa4f8a415c64c9d472c771a97703583b9 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Thu, 5 Nov 2026 00:10:15 +0100 Subject: [PATCH 4/5] docs: shorten the docstrings of the copy of a release Shorter docstrings for the code that this change adds or changes. Part of TNET-157. --- packages/timenet/src/timenet/client.py | 27 +++---- .../timenet/src/timenet/coordinator/plan.py | 74 ++++++------------- packages/timenet/src/timenet/engine/engine.py | 24 +++--- packages/timenet/src/timenet/errors.py | 25 +++---- packages/timenet/src/timenet/progress.py | 23 +++--- packages/timenet/src/timenet/registry/base.py | 23 +++--- .../src/timenet/registry/dependencies.py | 6 +- .../timenet/src/timenet/registry/local.py | 4 +- .../src/timenet/registry/remote/_download.py | 4 +- .../src/timenet/registry/remote/registry.py | 12 +-- .../timenet/src/timenet/registry/writable.py | 2 +- 11 files changed, 93 insertions(+), 131 deletions(-) diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index 1c20057d8..64d529a0e 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -308,18 +308,14 @@ def load( return reader.read(load=LoadPlan.ALL_EAGER) def _reject_unhosted_access(self, dataset_id: str, version: str | None) -> None: - """Raise for a non-open dataset read from a registry that does not host its data. - - Credentialed and restricted datasets cannot be redistributed, so TimeNet never serves their - bytes. Such a dataset is build-your-own: a local registry holds only what the user built, so a - local read is fine, but a hosted registry can only point the user at where to obtain access. + """Check dataset access against the registry's requirements. Args: - dataset_id: The bare dataset id. - version: The resolved version, or ``None`` for the latest. + dataset_id: The dataset ID without a version suffix. + version: The version string, or ``None`` for the latest. Raises: - TimeNetAccessError: If the dataset is not open and the registry requires open access. + TimeNetAccessError: If the dataset or its lineage lacks open access and the registry requires it. """ if not self._registry.requires_open_access: return @@ -394,20 +390,17 @@ def _missing_hint(*, auto_build: bool, fetch_from: BaseRegistry | None) -> str | def _prepare( registry: LocalRegistry, dataset_id: str, version: str | None, *, auto_build: bool, fetch_from: BaseRegistry | None ) -> DatasetVersion: - """Prepare a local version and its payload sources in the order that :meth:`TimeNet.open` names. + """Prepare a local release and its payload sources. - If a payload source is still missing after the copy from ``fetch_from``, this function builds it - with its own connector. That connector is the backend whose card declares the exact version of - the source. The version reads files of the source, so the build gets the pins of the version. - The build publishes the source only if it reproduces the pinned files. + Try local copies, then ``fetch_from``, then a build if ``auto_build`` is true. + A built payload source must match the files that the release pins. Returns: - The version with each payload source attached. + The release with each payload source attached. Raises: - TimeNetMissingParentError: If a payload source is absent and this function cannot copy or - build it. - TimeNetInvalidCardError: If the card of a connector cannot be read. + TimeNetMissingParentError: If a payload source is missing and cannot be copied or built. + TimeNetInvalidCardError: If a connector's dataset card cannot be read. """ # ruff: ignore[docstring-extraneous-exception] - raised by _obtain and exact_builder try: manifest = registry.get_manifest(dataset_id, version) diff --git a/packages/timenet/src/timenet/coordinator/plan.py b/packages/timenet/src/timenet/coordinator/plan.py index f1ed9d4b3..0cdc59012 100644 --- a/packages/timenet/src/timenet/coordinator/plan.py +++ b/packages/timenet/src/timenet/coordinator/plan.py @@ -198,52 +198,29 @@ def plan_parents( build_parents: bool = False, force_build: bool = False, ) -> list[ParentStep]: - """Plan how to get each parent release that a build of ``dataset_id`` needs. - - For each declared parent, a build needs the parent and the payload sources of that parent. It - does not need the payload sources of those payload sources. The plan visits the declared parents - of a release only if the plan builds that release. So if the output already holds the requested - release, the plan visits its parents only if ``force_build`` is set. If ``force_build`` is not - set, the plan visits only the payload sources of that release. - - If the output does not hold a release, the plan fetches it from ``fetch_from``. If ``fetch_from`` - does not hold it either, the plan builds it with the installed connector that declares that - exact version. If a local output holds a damaged copy, the plan fetches the release again and - does not build it. A registry that is not local holds only releases whose cards declare - ``access: open``. So for such an output, each release that the plan visits must declare - ``access: open``. If ``fetch_from`` is not local, each release that the plan fetches from it - must declare ``access: open`` too. - - The plan works in two phases. First, a walk from the requested release finds the graph of the - releases that the build needs. Then :class:`graphlib.TopologicalSorter` orders the graph and - finds cycles, and the plan makes its steps in that order. + """Plan how to fetch or build the releases a dataset needs. + + Prefer complete copies, then ``fetch_from``, then builds if ``build_parents`` is true. + An existing release needs only its payload sources unless ``force_build`` is set. Args: dataset_id: The dataset to build. - output: The registry that the build writes to. - fetch_from: The registry to fetch missing parents from. If it is ``None``, the plan fetches - nothing. + output: The registry to build into. + fetch_from: The registry to fetch missing or damaged parents from, or ``None``. build_parents: Build missing parents with their installed connectors. - force_build: Build the requested release again, even if the output holds it, and plan its - parents. + force_build: Plan the requested release's parents even if the output already holds the release. Returns: - The steps to fetch or build parents. Each step comes after the steps of the releases that it - depends on. + Fetch or build steps, ordered with dependencies first. Raises: - LookupError: If no installed connector declares a card for ``dataset_id``. - TimeNetInvalidCardError: If the card of an installed connector cannot be read. - TimeFValidationError: If the parents form a cycle, if an installed connector declares another - version than a parent pins, or if ``fetch_from`` is set and the output is not local. - TimeNetMissingParentError: If a parent or a payload source is missing and ``build_parents`` - is not set. + LookupError: If no installed connector declares a dataset card for ``dataset_id``. + TimeNetInvalidCardError: If an installed connector's dataset card cannot be read. + TimeFValidationError: If parents cycle, a connector builds the wrong version, or fetching uses non-local output. + TimeNetMissingParentError: If a parent or payload source is missing and ``build_parents`` is false. TimeNetDatasetNotFoundError: If a parent is missing and no installed connector builds it. - TimeNetFileError: If a local output holds a damaged copy of a parent, and the plan cannot - fetch the parent again. - TimeNetAccessError: If the output is not local and the plan visits a release whose card, or - a card in its lineage, does not declare ``access: open``. Also if ``fetch_from`` is not - local and the plan would fetch such a release from it. + TimeNetFileError: If a local parent copy has a missing or resized file and cannot be fetched again. + TimeNetAccessError: If a release or its lineage lacks open access where a registry requires it. """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers of the walk if fetch_from is not None and not isinstance(output, LocalRegistry): # A registry that is not local accepts a child only if it already holds the parents. Also, a @@ -346,21 +323,16 @@ def _locate( ) -> tuple[Manifest | None, Literal["fetch"] | None]: # The damage is raised from a variable, the access error by require_open_access. # ruff: ignore[docstring-extraneous-exception] - """Find a release in the output, or else in ``fetch_from``. + """Find a release in ``output`` or ``fetch_from``. - The output must hold an intact copy. In a local output, a copy with a missing file or a file of - another size is damaged. + A local copy is intact when every file has its recorded size. Returns: - The manifest and ``None`` if the output holds an intact copy. The manifest and ``"fetch"`` if - the plan fetches the release. ``None`` and ``None`` if the plan builds it. + ``(manifest, None)`` for an output copy, ``(manifest, "fetch")`` for a fetch, or ``(None, None)`` if absent. Raises: - TimeNetFileError: If a local output holds a copy with a missing file or a file of another - size, and the plan cannot fetch the release again. - TimeNetAccessError: If the card of the release, or a card in its lineage, does not declare - ``access: open``, and the output that holds the release, or the ``fetch_from`` registry - to fetch it from, is not local. + TimeNetFileError: If the local copy has a missing or resized file and ``fetch_from`` lacks the release. + TimeNetAccessError: If the release or its lineage lacks open access where its registry requires it. """ manifest = _manifest(output, reference) damage = None @@ -399,14 +371,14 @@ def _card_to_build( # ruff: ignore[too-many-arguments] pinned: bool, searched: str, ) -> DatasetMetadata: - """Return the card to build a release that the output does not hold and the plan cannot fetch. + """Find a dataset card for building a missing release. Returns: - The card of the installed connector that builds exactly ``reference``. + The installed connector's card for the exact release in ``reference``. Raises: - TimeNetMissingParentError: If ``build_parents`` is not set. - TimeNetAccessError: If the output is not local and the card does not declare ``access: open``. + TimeNetMissingParentError: If ``build_parents`` is false. + TimeNetAccessError: If the card lacks open access and the output requires it. """ # ruff: ignore[docstring-extraneous-exception] - the access error is raised by require_open_access if not build_parents: raise TimeNetMissingParentError(reference, needed_by, searched, payload_source=pinned) diff --git a/packages/timenet/src/timenet/engine/engine.py b/packages/timenet/src/timenet/engine/engine.py index e46264409..f34c8d339 100644 --- a/packages/timenet/src/timenet/engine/engine.py +++ b/packages/timenet/src/timenet/engine/engine.py @@ -113,26 +113,26 @@ def publish_pipeline( # ruff: ignore[too-many-arguments] force: bool = False, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> str: - """Run a connector and publish its dataset to a registry. + """Build and publish a dataset to a writable registry. - If the version already exists, return it unless ``force`` is set. Remote and S3 registries - require ``access: open`` on the dataset card and its lineage. + Reuse an existing version unless ``force`` is set. + The dataset card and lineage must declare open access if the registry requires it. Args: connector: The connector to build. - registry: The destination registry. - cache_dir: Download directory, or ``None`` for ``/``. - keep_cache: Keep the default cache after publishing (a supplied ``cache_dir`` is always kept). - progress_cb: Optional callback for write progress. - force: Republish even if the version already exists. - pinned_by: Releases that already read this release's files, with the pins that the new files must match. + registry: The writable registry to publish to. + cache_dir: The download directory, or ``None`` for ``/``. + keep_cache: Keep the default cache after publishing. A supplied ``cache_dir`` is always kept. + progress_cb: Callback for writer progress events, or ``None``. + force: Publish again even if the version already exists. + pinned_by: Releases that read this release's files, mapped to file records the build must match. Returns: - The published version string. + The existing or newly published version string. Raises: - TimeFValidationError: If declared parents lack a custom ``compose``, or ``compose`` changes the dataset card. - TimeNetAccessError: If a nonlocal publish includes a dataset or ancestor without open access. + TimeFValidationError: If a connector declares parents without overriding ``compose``, or changes its card. + TimeNetAccessError: If the card, dataset, or lineage lacks open access and the registry requires it. """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_composes, _compose, and require_open_access values_backend = _check_values_backend(connector) metadata = connector.metadata() diff --git a/packages/timenet/src/timenet/errors.py b/packages/timenet/src/timenet/errors.py index 427896e24..e43f01108 100644 --- a/packages/timenet/src/timenet/errors.py +++ b/packages/timenet/src/timenet/errors.py @@ -34,15 +34,14 @@ class TimeNetDatasetNotFoundError(TimeNetError): class TimeNetMissingParentError(TimeNetDatasetNotFoundError): - """A release that another release needs is absent: a declared parent of a build, or a payload source. + """A required parent release or payload source is missing. Attributes: missing: The absent release. needed_by: The release that needs it. - payload_source: Whether ``needed_by`` reads files of ``missing`` (a payload source), rather than - declaring it as a parent to build from. - searched: Where the release was looked for. - hint: How the caller can obtain the release, added by the front end, or ``None``. + payload_source: ``True`` if ``needed_by`` reads files of ``missing``, ``False`` for a declared build parent. + searched: The search location. + hint: Advice on obtaining the release, or ``None``. """ def __init__(self, missing: "DatasetRef", needed_by: "DatasetRef", searched: str, *, payload_source: bool) -> None: @@ -65,18 +64,16 @@ def __str__(self) -> str: class TimeNetVersionConflictError(TimeNetError): - """A registry already holds a release with other files than the copy that would go there. + """A registry already holds a release with other files than the copy. - A published version cannot change, so other files need a new version. + Different files require a new version. Attributes: - release: The release that the registry holds with other files. - registry: The registry that holds it, as text. - source: The registry that the copy comes from, as text. - needed_by: The release that the copy is for, if ``release`` is a payload source that it - needs, directly or through another payload source. ``None`` if ``release`` is the - release that the copy is for. - hint: What the caller can do instead, added by the front end, or ``None``. + release: The release with conflicting files. + registry: The destination registry, as text. + source: The source registry, as text. + needed_by: The requested release that needs this payload source, or ``None`` if ``release`` is the request. + hint: Advice to append to the error message, or ``None``. """ def __init__( diff --git a/packages/timenet/src/timenet/progress.py b/packages/timenet/src/timenet/progress.py index eb1659ef6..04a399754 100644 --- a/packages/timenet/src/timenet/progress.py +++ b/packages/timenet/src/timenet/progress.py @@ -1,8 +1,7 @@ -"""Typed progress events for copies and builds. +"""Progress events for copying and building releases. -A function that copies or builds releases sends these events to an optional -:data:`ProgressCallback` and prints nothing. Each step, the copy or the build of one release, sends -:class:`StepStarted` first and :class:`StepFinished` only if it succeeds. +Each release starts with a :class:`StepStarted` event. +A :class:`StepFinished` event follows only if the copy or build succeeds. """ from collections.abc import Callable @@ -14,11 +13,11 @@ @dataclass(frozen=True) class StepStarted: - """A step started to copy or build one release. + """The start of a copy or build for one release. Attributes: - release: The release that the step copies or builds. - action: ``"copy"`` or ``"build"``. + release: The release to copy or build. + action: The operation, either ``"copy"`` or ``"build"``. """ release: DatasetRef @@ -27,11 +26,11 @@ class StepStarted: @dataclass(frozen=True) class StepFinished: - """A step finished the copy or the build of one release. + """The successful completion of a copy or build for one release. Attributes: - release: The release that the step copied or built. - action: ``"copy"`` or ``"build"``. + release: The copied or built release. + action: The completed operation, either ``"copy"`` or ``"build"``. """ release: DatasetRef @@ -39,7 +38,7 @@ class StepFinished: ProgressEvent = StepStarted | StepFinished -"""One progress event.""" +"""An event that marks the start or successful end of a copy or build.""" ProgressCallback = Callable[[ProgressEvent], None] -"""A function that receives each progress event.""" +"""A callback that receives a progress event and returns nothing.""" diff --git a/packages/timenet/src/timenet/registry/base.py b/packages/timenet/src/timenet/registry/base.py index 012d52628..d0d931fbc 100644 --- a/packages/timenet/src/timenet/registry/base.py +++ b/packages/timenet/src/timenet/registry/base.py @@ -32,7 +32,7 @@ class BaseRegistry(ABC): - """A source of TimeF datasets: their manifests, files, and searchable metadata.""" + """A registry of dataset manifests, files, and searchable metadata.""" def __str__(self) -> str: return type(self).__name__ @@ -44,7 +44,7 @@ def local_root(self) -> Path: @property def requires_open_access(self) -> bool: - """Whether this registry holds and serves only releases with open access.""" + """Whether stored or served releases must declare open access.""" return True @abstractmethod @@ -158,22 +158,21 @@ def download_release( # ruff: ignore[too-many-arguments] manifest: Manifest | None = None, progress_cb: ProgressCallback | None = None, ) -> None: - """Download the files of one release into a local directory. + """Download one release's files to a local directory. - The download replaces ``directory`` only after every file arrived and matched its manifest - record, so a failed download leaves no partial copy. + Reuse a complete copy unless ``force`` is set. + Replace ``directory`` only after every downloaded file matches its manifest record. Args: dataset_id: The dataset ID. - version: The version string. - directory: The directory of the release, ``//``. - force: Download again, even if ``directory`` holds a complete copy. - manifest: The manifest of the release, or ``None`` to read it from this registry. - progress_cb: Receives the size of each downloaded file, or ``None``. + version: The exact version string. + directory: The local release directory, ``//``. + force: Download again even if a complete copy exists. + manifest: The release's manifest, or ``None`` to fetch it from this registry. + progress_cb: Callback that receives each downloaded file's byte count, or ``None``. Raises: - TimeNetFileError: If a downloaded file has another size or checksum than its manifest - records. + TimeNetFileError: If a downloaded file's size or checksum differs from its manifest record. """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_staged_file if manifest is None: manifest = self.get_manifest(dataset_id, version) diff --git a/packages/timenet/src/timenet/registry/dependencies.py b/packages/timenet/src/timenet/registry/dependencies.py index d4c46cc17..1eb026535 100644 --- a/packages/timenet/src/timenet/registry/dependencies.py +++ b/packages/timenet/src/timenet/registry/dependencies.py @@ -431,10 +431,12 @@ def _download_dataset_version( # ruff: ignore[too-many-arguments] progress_cb: Callable[[int], None] | None, before_fetch: Callable[[Manifest], object] | None = None, ) -> Manifest: - """Fetch one complete release, checking its identity before downloading its files. + """Check a release's identity and download its files. + + Run ``before_fetch`` after checking the manifest and before downloading files. Returns: - The manifest after the registry has verified and stored the downloaded files. + The manifest of the downloaded and verified release. """ if manifest is None: manifest = registry.get_manifest(reference.dataset_id, str(reference.version)) diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py index 54979ab29..57435150b 100644 --- a/packages/timenet/src/timenet/registry/local.py +++ b/packages/timenet/src/timenet/registry/local.py @@ -50,10 +50,10 @@ def local_root(self) -> Path: @property def requires_open_access(self) -> bool: - """Whether this registry holds and serves only releases with open access. + """Whether local releases must declare open access. Returns: - ``False``: a local registry holds releases of any access. + ``False``, since a local registry accepts any access level. """ return False diff --git a/packages/timenet/src/timenet/registry/remote/_download.py b/packages/timenet/src/timenet/registry/remote/_download.py index 2006933d8..12b62e250 100644 --- a/packages/timenet/src/timenet/registry/remote/_download.py +++ b/packages/timenet/src/timenet/registry/remote/_download.py @@ -52,13 +52,13 @@ def download_version_files( # ruff: ignore[too-many-arguments] Args: http: The registry HTTP client. - manifest: The version's parsed manifest (its file list drives the download). + manifest: The release's manifest. dataset_id: The ``org/name`` id. version: The version string. dest_dir: The target ``<...>//`` directory. force: Re-download even if ``dest_dir/manifest.json`` already exists. max_concurrency: Maximum concurrent file downloads. - progress_cb: Called with each chunk's byte count as it is written, for a progress display. + progress_cb: Callback that receives each downloaded chunk's byte count, or ``None``. Returns: ``dest_dir``. diff --git a/packages/timenet/src/timenet/registry/remote/registry.py b/packages/timenet/src/timenet/registry/remote/registry.py index 5cbc5de96..94b04ce7a 100644 --- a/packages/timenet/src/timenet/registry/remote/registry.py +++ b/packages/timenet/src/timenet/registry/remote/registry.py @@ -140,15 +140,15 @@ def download_release( # ruff: ignore[too-many-arguments] manifest: Manifest | None = None, progress_cb: ProgressCallback | None = None, ) -> None: - """Download the files of one release into a local directory. + """Download one release's files to a local directory. Args: dataset_id: The dataset ID, as ``org/name``. - version: The version string. - directory: The directory of the release, ``//``. - force: Download again, even if ``directory`` holds a complete copy. - manifest: The manifest of the release, or ``None`` to read it from this registry. - progress_cb: Receives the size of each downloaded chunk, or ``None``. + version: The exact version string. + directory: The local release directory, ``//``. + force: Download again even if a complete copy exists. + manifest: The release's manifest, or ``None`` to fetch it from this registry. + progress_cb: Callback that receives each downloaded chunk's byte count, or ``None``. """ if manifest is None: manifest = self.get_manifest(dataset_id, version) diff --git a/packages/timenet/src/timenet/registry/writable.py b/packages/timenet/src/timenet/registry/writable.py index 953ca548c..111990b3c 100644 --- a/packages/timenet/src/timenet/registry/writable.py +++ b/packages/timenet/src/timenet/registry/writable.py @@ -91,7 +91,7 @@ def _read_release(directory: Path) -> Manifest: return manifest def exists(self, dataset_id: str, version: str) -> bool: - """Return whether a committed version already exists in this registry. + """Check whether this registry holds a dataset version. Args: dataset_id: The dataset id. From 3e3044759b55ed8bedcff0de40c18b9287b019a6 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Thu, 5 Nov 2026 13:55:29 +0100 Subject: [PATCH 5/5] fix(registry): refuse changed cards, keep copies when a download fails A copy compared only the file records with the copy in the destination. So a rebuild with the same files and another card or lineage counted as the same release, and the destination kept the old card. Now the copy compares the whole manifest without its build environment. TimeNetVersionConflictError names the difference: "files" or "card". A forced download deleted the old copy before it renamed the new one in place. If the rename failed, no copy was left. Both download paths now stage the files with staged_directory(), which keeps the old copy until the new one is in place. download_release() takes a DatasetRef, as copy_release() does. The registries get find_manifest(), which returns None for a release that a registry does not hold, and LocalRegistry.damage(), which names the first missing or resized file. The copy uses both instead of its own checks. Part of TNET-157. --- packages/timenet/src/timenet/_fs.py | 27 ++++++++++ packages/timenet/src/timenet/errors.py | 24 ++++++--- .../timenet/src/timenet/manifest/__init__.py | 3 +- .../timenet/src/timenet/manifest/manifest.py | 13 +++++ packages/timenet/src/timenet/registry/base.py | 54 ++++++++++--------- .../src/timenet/registry/dependencies.py | 9 +--- .../timenet/src/timenet/registry/local.py | 36 +++++++++---- .../src/timenet/registry/remote/_download.py | 17 +----- .../src/timenet/registry/remote/registry.py | 9 ++-- .../timenet/src/timenet/registry/writable.py | 7 +-- .../timenet/tests/registry/test_registry.py | 21 ++++++++ packages/timenet/tests/test_errors.py | 18 ++++--- 12 files changed, 154 insertions(+), 84 deletions(-) diff --git a/packages/timenet/src/timenet/_fs.py b/packages/timenet/src/timenet/_fs.py index ec216e113..97c7bf03e 100644 --- a/packages/timenet/src/timenet/_fs.py +++ b/packages/timenet/src/timenet/_fs.py @@ -1,5 +1,7 @@ """Replace a directory with a complete staging directory.""" +from collections.abc import Generator +from contextlib import contextmanager from pathlib import Path import shutil import uuid @@ -37,3 +39,28 @@ def publish_directory(staging: Path, final: Path) -> None: if old is not None: # The new directory is published, so failing to delete the old one must not fail the publish. shutil.rmtree(old, ignore_errors=True) + + +@contextmanager +def staged_directory(final: Path) -> Generator[Path]: + """Yield an empty staging directory beside ``final``, and publish it as ``final`` if the block succeeds. + + The old ``final`` stays in place until the new directory is complete. The staging directory is + removed in every case. + + Args: + final: The destination directory. + + Yields: + The staging directory. + """ + final.parent.mkdir(parents=True, exist_ok=True) + # A hard kill can leave a staging directory behind. Do not sweep other .tmp-* directories: + # a concurrent write of the same release may still use one. + staging = final.with_name(f"{final.name}.tmp-{uuid.uuid4().hex}") + staging.mkdir() + try: + yield staging + publish_directory(staging, final) + finally: + shutil.rmtree(staging, ignore_errors=True) diff --git a/packages/timenet/src/timenet/errors.py b/packages/timenet/src/timenet/errors.py index e43f01108..f87b91f26 100644 --- a/packages/timenet/src/timenet/errors.py +++ b/packages/timenet/src/timenet/errors.py @@ -13,7 +13,7 @@ """ from functools import partial -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Literal if TYPE_CHECKING: @@ -64,33 +64,41 @@ def __str__(self) -> str: class TimeNetVersionConflictError(TimeNetError): - """A registry already holds a release with other files than the copy. + """A registry already holds a release with other files or another card than the copy. - Different files require a new version. + A changed release needs a new version. Attributes: - release: The release with conflicting files. + release: The release that differs. registry: The destination registry, as text. source: The source registry, as text. needed_by: The requested release that needs this payload source, or ``None`` if ``release`` is the request. + difference: ``"files"`` if the file records differ, else ``"card"`` for the rest of the manifest. hint: Advice to append to the error message, or ``None``. """ def __init__( - self, release: "DatasetRef", registry: str, source: str, needed_by: "DatasetRef | None" = None + self, + release: "DatasetRef", + registry: str, + source: str, + needed_by: "DatasetRef | None" = None, + difference: Literal["files", "card"] = "files", ) -> None: - super().__init__(release, registry, source, needed_by) + super().__init__(release, registry, source, needed_by, difference) self.release = release self.registry = registry self.source = source self.needed_by = needed_by + self.difference = difference self.hint: str | None = None def __str__(self) -> str: held = str(self.release) if self.needed_by is None else f"{self.release}, a payload source of {self.needed_by}," + other = "other files" if self.difference == "files" else "another card" message = ( - f"{self.registry} already holds {held} with other files than {self.source}. " - "A published version cannot change, so other files need a new version." + f"{self.registry} already holds {held} with {other} than {self.source}. " + "A published version cannot change, so a changed release needs a new version." ) return message if self.hint is None else f"{message} {self.hint}" diff --git a/packages/timenet/src/timenet/manifest/__init__.py b/packages/timenet/src/timenet/manifest/__init__.py index 480bdf788..9f8984f9a 100644 --- a/packages/timenet/src/timenet/manifest/__init__.py +++ b/packages/timenet/src/timenet/manifest/__init__.py @@ -2,7 +2,7 @@ from timenet.manifest.counts import ManifestCounts from timenet.manifest.files import FilePart, ManifestFiles -from timenet.manifest.manifest import BuildEnvironment, Manifest, parse_manifest, read_manifest +from timenet.manifest.manifest import BuildEnvironment, Manifest, parse_manifest, read_manifest, same_release from timenet.manifest.payload_source import PayloadSource @@ -15,4 +15,5 @@ "PayloadSource", "parse_manifest", "read_manifest", + "same_release", ] diff --git a/packages/timenet/src/timenet/manifest/manifest.py b/packages/timenet/src/timenet/manifest/manifest.py index f93815e71..fb1137ff9 100644 --- a/packages/timenet/src/timenet/manifest/manifest.py +++ b/packages/timenet/src/timenet/manifest/manifest.py @@ -106,6 +106,19 @@ def to_json(self) -> str: return validated.model_dump_json(indent=2, by_alias=True) +def same_release(a: Manifest, b: Manifest) -> bool: + """Compare two manifests, ignoring their build environments. + + Args: + a: One manifest. + b: The other manifest. + + Returns: + Whether the manifests are equal apart from ``build_env``. + """ + return a.model_copy(update={"build_env": {}}) == b.model_copy(update={"build_env": {}}) + + def parse_manifest(text: str | bytes, where: str) -> Manifest: """Parse the JSON text of a stored manifest. diff --git a/packages/timenet/src/timenet/registry/base.py b/packages/timenet/src/timenet/registry/base.py index d0d931fbc..afbea2b7e 100644 --- a/packages/timenet/src/timenet/registry/base.py +++ b/packages/timenet/src/timenet/registry/base.py @@ -11,13 +11,14 @@ from pathlib import Path import shutil from typing import BinaryIO, TypeVar -import uuid -from timenet.errors import TimeFValidationError, TimeNetFileError +from timenet._fs import staged_directory +from timenet.errors import TimeFValidationError, TimeNetDatasetNotFoundError, TimeNetFileError from timenet.format.checksums import file_checksum +from timenet.format.constants import MANIFEST_FILE from timenet.manifest import FilePart, Manifest from timenet.registry._paths import safe_version_path, split_release_dir -from timenet.registry.dependencies import prepare_local_copy +from timenet.registry.dependencies import prepare_local_copy, validate_manifest_reference from timenet.registry.shared import complete_copy from timenet.registry.version import DatasetVersion from timenet.types import DatasetMetadata, Domain, License, Task @@ -70,6 +71,25 @@ def get_manifest(self, dataset_id: str, version: str | None = None) -> Manifest: TimeNetDatasetNotFoundError: If the dataset id or version is unknown. """ + def find_manifest(self, release: DatasetRef) -> Manifest | None: + """Return the manifest of a release, or ``None`` if this registry does not hold it. + + Args: + release: The exact release. + + Returns: + The manifest, or ``None``. + + Raises: + TimeFFormatError: If the stored manifest names another release. + """ # ruff: ignore[docstring-extraneous-exception] - raised by validate_manifest_reference + try: + manifest = self.get_manifest(release.dataset_id, str(release.version)) + except TimeNetDatasetNotFoundError: + return None + validate_manifest_reference(manifest, release) + return manifest + @abstractmethod def open_file(self, dataset_id: str, version: str, relpath: str) -> BinaryIO: """Open one file of a dataset version for binary reading. @@ -148,10 +168,9 @@ def download_version( # ruff: ignore[too-many-arguments] progress_cb=progress_cb, ) - def download_release( # ruff: ignore[too-many-arguments] + def download_release( self, - dataset_id: str, - version: str, + reference: DatasetRef, directory: str | Path, *, force: bool = False, @@ -164,8 +183,7 @@ def download_release( # ruff: ignore[too-many-arguments] Replace ``directory`` only after every downloaded file matches its manifest record. Args: - dataset_id: The dataset ID. - version: The exact version string. + reference: The exact release. directory: The local release directory, ``//``. force: Download again even if a complete copy exists. manifest: The release's manifest, or ``None`` to fetch it from this registry. @@ -174,31 +192,19 @@ def download_release( # ruff: ignore[too-many-arguments] Raises: TimeNetFileError: If a downloaded file's size or checksum differs from its manifest record. """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_staged_file + dataset_id, version = reference.dataset_id, str(reference.version) if manifest is None: manifest = self.get_manifest(dataset_id, version) target = Path(directory) if complete_copy(target, manifest) and not force: return # already downloaded and current - parent = target.parent - parent.mkdir(parents=True, exist_ok=True) - # Each download stages into a unique dir and removes it in the finally below. A hard kill - # (SIGKILL/power loss) can leave one behind, but do NOT sweep sibling .tmp-* dirs here: - # a concurrent download of the same version has a live staging dir with the same prefix, and - # sweeping it would break that download mid-write. A rare orphaned dir is the lesser evil. - staging = parent / f"{target.name}.tmp-{uuid.uuid4().hex}" - release = DatasetRef.parse(f"{dataset_id}@{version}") - try: + with staged_directory(target) as staging: for part in manifest.files.all_files(): self._stage_file(dataset_id, version, part.path, staging) - self._check_staged_file(staging / part.path, part, release) + self._check_staged_file(staging / part.path, part, reference) if progress_cb is not None: progress_cb(part.size) - (staging / "manifest.json").write_text(manifest.to_json()) - if target.exists(): - shutil.rmtree(target) - staging.replace(target) - finally: - shutil.rmtree(staging, ignore_errors=True) + (staging / MANIFEST_FILE).write_text(manifest.to_json()) def _check_staged_file(self, path: Path, part: FilePart, release: DatasetRef) -> None: """Compare one downloaded file with its manifest record, first by size, then by checksum. diff --git a/packages/timenet/src/timenet/registry/dependencies.py b/packages/timenet/src/timenet/registry/dependencies.py index 1eb026535..29b533bd4 100644 --- a/packages/timenet/src/timenet/registry/dependencies.py +++ b/packages/timenet/src/timenet/registry/dependencies.py @@ -443,14 +443,7 @@ def _download_dataset_version( # ruff: ignore[too-many-arguments] validate_manifest_reference(manifest, reference) if before_fetch is not None: before_fetch(manifest) - registry.download_release( - reference.dataset_id, - str(reference.version), - directory, - force=True, - manifest=manifest, - progress_cb=progress_cb, - ) + registry.download_release(reference, directory, force=True, manifest=manifest, progress_cb=progress_cb) return manifest diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py index 57435150b..0376d8ab0 100644 --- a/packages/timenet/src/timenet/registry/local.py +++ b/packages/timenet/src/timenet/registry/local.py @@ -4,15 +4,16 @@ from pathlib import Path import shutil from typing import BinaryIO -import uuid -from timenet._fs import publish_directory +import pyarrow.fs as pafs + +from timenet._fs import staged_directory from timenet.dataset import TimeFDataset -from timenet.errors import TimeFFormatError, TimeNetDatasetNotFoundError +from timenet.errors import TimeFFormatError, TimeNetDatasetNotFoundError, TimeNetFileError from timenet.format.constants import MANIFEST_FILE from timenet.manifest import Manifest, PayloadSource, read_manifest from timenet.registry._paths import release_dir -from timenet.registry.dependencies import open_local_release +from timenet.registry.dependencies import check_file_sizes, open_local_release from timenet.registry.version import DatasetVersion from timenet.registry.writable import WritableRegistry from timenet.types import DatasetMetadata, Version @@ -213,19 +214,32 @@ def put_release(self, from_directory: Path) -> DatasetRef: """ # ruff: ignore[docstring-extraneous-exception] - raised by _read_release and publish_directory manifest = self._read_release(from_directory) reference = DatasetRef(dataset_id=manifest.dataset_id, version=manifest.metadata.dataset_version) - final = release_dir(self._root, reference) - final.parent.mkdir(parents=True, exist_ok=True) - staging = final.with_name(f"{final.name}.tmp-{uuid.uuid4().hex}") - try: + with staged_directory(release_dir(self._root, reference)) as staging: for relpath in (*sorted({part.path for part in manifest.files.all_files()}), MANIFEST_FILE): target = staging / relpath target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(from_directory / relpath, target) - publish_directory(staging, final) - finally: - shutil.rmtree(staging, ignore_errors=True) return reference + def damage(self, release: DatasetRef, manifest: Manifest | None = None) -> TimeNetFileError | None: + """Return the first missing or resized file of a stored release, or ``None`` for a complete copy. + + Args: + release: The exact release. + manifest: Its manifest, or ``None`` to read the stored one. + + Returns: + The error that names the file, or ``None``. + """ + if manifest is None: + manifest = self.get_manifest(release.dataset_id, str(release.version)) + version = DatasetVersion(manifest, pafs.LocalFileSystem(), str(release_dir(self._root, release))) + try: + check_file_sizes(version, manifest.files.all_files(), owner=str(release)) + except TimeNetFileError as error: + return error + return None + def open_version(self, dataset_id: str, version: str | None = None) -> DatasetVersion: """Open a committed version as a local, random-access handle. diff --git a/packages/timenet/src/timenet/registry/remote/_download.py b/packages/timenet/src/timenet/registry/remote/_download.py index 12b62e250..093a7089b 100644 --- a/packages/timenet/src/timenet/registry/remote/_download.py +++ b/packages/timenet/src/timenet/registry/remote/_download.py @@ -11,13 +11,12 @@ from collections.abc import Coroutine import hashlib from pathlib import Path -import shutil import threading from typing import Any -import uuid import httpx +from timenet._fs import staged_directory from timenet.errors import TimeNetFileError from timenet.manifest import Manifest from timenet.manifest.files import FilePart @@ -67,21 +66,9 @@ def download_version_files( # ruff: ignore[too-many-arguments] if complete_copy(dest_dir, manifest) and not force: return dest_dir files = list(manifest.files.all_files()) - parent = dest_dir.parent - parent.mkdir(parents=True, exist_ok=True) - # Each download stages into a unique dir and removes it in the finally below. A hard kill - # (SIGKILL/power loss) can leave one behind, but do NOT sweep sibling .tmp-* dirs here: a - # concurrent download of the same version has a live staging dir with the same prefix, and sweeping - # it would break that download mid-write. A rare orphaned dir is the lesser evil. - staging = parent / f"{dest_dir.name}.tmp-{uuid.uuid4().hex}" - try: + with staged_directory(dest_dir) as staging: _run(_download_all(http, dataset_id, version, files, staging, max_concurrency, progress_cb)) (staging / "manifest.json").write_text(manifest.to_json()) # already parsed; no extra round trip - if dest_dir.exists(): - shutil.rmtree(dest_dir) - staging.replace(dest_dir) - finally: - shutil.rmtree(staging, ignore_errors=True) return dest_dir diff --git a/packages/timenet/src/timenet/registry/remote/registry.py b/packages/timenet/src/timenet/registry/remote/registry.py index 94b04ce7a..258363079 100644 --- a/packages/timenet/src/timenet/registry/remote/registry.py +++ b/packages/timenet/src/timenet/registry/remote/registry.py @@ -130,10 +130,9 @@ def _open_remote_version(self, reference: DatasetRef) -> DatasetVersion: filesystem = RegistryFileSystem(self._base_url, files, token=self._http.token, transport=self._http.transport) return DatasetVersion(manifest, pafs.PyFileSystem(pafs.FSSpecHandler(filesystem)), root) - def download_release( # ruff: ignore[too-many-arguments] + def download_release( self, - dataset_id: str, - version: str, + reference: DatasetRef, directory: str | Path, *, force: bool = False, @@ -143,13 +142,13 @@ def download_release( # ruff: ignore[too-many-arguments] """Download one release's files to a local directory. Args: - dataset_id: The dataset ID, as ``org/name``. - version: The exact version string. + reference: The exact release. directory: The local release directory, ``//``. force: Download again even if a complete copy exists. manifest: The release's manifest, or ``None`` to fetch it from this registry. progress_cb: Callback that receives each downloaded chunk's byte count, or ``None``. """ + dataset_id, version = reference.dataset_id, str(reference.version) if manifest is None: manifest = self.get_manifest(dataset_id, version) download_version_files( diff --git a/packages/timenet/src/timenet/registry/writable.py b/packages/timenet/src/timenet/registry/writable.py index 111990b3c..a4d434b6f 100644 --- a/packages/timenet/src/timenet/registry/writable.py +++ b/packages/timenet/src/timenet/registry/writable.py @@ -13,7 +13,6 @@ import pyarrow.fs as pafs from timenet.dataset import TimeFDataset -from timenet.errors import TimeNetDatasetNotFoundError from timenet.format.constants import MANIFEST_FILE from timenet.manifest import Manifest, PayloadSource, read_manifest from timenet.registry.base import BaseRegistry @@ -100,8 +99,4 @@ def exists(self, dataset_id: str, version: str) -> bool: Returns: ``True`` if the version has a committed manifest, else ``False``. """ - try: - self.get_manifest(dataset_id, version) - except TimeNetDatasetNotFoundError: - return False - return True + return self.find_manifest(DatasetRef.parse(f"{dataset_id}@{version}")) is not None diff --git a/packages/timenet/tests/registry/test_registry.py b/packages/timenet/tests/registry/test_registry.py index 00cf716f8..01e1438f4 100644 --- a/packages/timenet/tests/registry/test_registry.py +++ b/packages/timenet/tests/registry/test_registry.py @@ -152,6 +152,27 @@ def test_get_manifest_names_the_path_of_a_cut_manifest(registry_root): LocalRegistry(registry_root).get_manifest("timenet/hello-world", "1.0.0") +def test_find_manifest_answers_none_for_an_absent_release_and_refuses_a_misplaced_one(registry_root): + registry = LocalRegistry(registry_root) + assert registry.find_manifest(DatasetRef.parse("timenet/hello-world@9.0.0")) is None + misplaced = registry_root / "timenet/hello-world/2.0.0" + misplaced.mkdir() + (misplaced / "manifest.json").write_bytes((registry_root / "timenet/hello-world/1.0.0/manifest.json").read_bytes()) + with pytest.raises(TimeFFormatError, match="declares a different dataset ID or version"): + registry.find_manifest(DatasetRef.parse("timenet/hello-world@2.0.0")) + + +def test_damage_names_the_missing_file_of_a_stored_release(registry_root): + registry = LocalRegistry(registry_root) + release = DatasetRef.parse("timenet/hello-world@1.0.0") + assert registry.damage(release) is None + part = registry.get_manifest(release.dataset_id, "1.0.0").files.control[0] + (registry_root / "timenet/hello-world/1.0.0" / part.path).unlink() + damage = registry.damage(release) + assert damage is not None + assert (damage.release, damage.path) == (release, part.path) + + def test_latest_version_ignores_staging_dirs(registry_root): # A crashed build can leave a `.tmp-` sibling (briefly holding a manifest.json). stale = registry_root / "demo/ecg" / "2.0.0.tmp-deadbeef" diff --git a/packages/timenet/tests/test_errors.py b/packages/timenet/tests/test_errors.py index 921bc5d6c..075de37d1 100644 --- a/packages/timenet/tests/test_errors.py +++ b/packages/timenet/tests/test_errors.py @@ -116,15 +116,21 @@ def test_catchable_as_base(): ), ( TimeNetVersionConflictError(_A, "registry here", "registry there"), - {"release": _A, "registry": "registry here", "source": "registry there", "needed_by": None}, + { + "release": _A, + "registry": "registry here", + "source": "registry there", + "needed_by": None, + "difference": "files", + }, "registry here already holds org/a@1.0.0 with other files than registry there. " - "A published version cannot change, so other files need a new version.", + "A published version cannot change, so a changed release needs a new version.", ), ( - TimeNetVersionConflictError(_A, "registry here", "registry there", _B), - {"release": _A, "needed_by": _B}, - "registry here already holds org/a@1.0.0, a payload source of org/b@2.0.0, with other files than " - "registry there. A published version cannot change, so other files need a new version.", + TimeNetVersionConflictError(_A, "registry here", "registry there", _B, "card"), + {"release": _A, "needed_by": _B, "difference": "card"}, + "registry here already holds org/a@1.0.0, a payload source of org/b@2.0.0, with another card than " + "registry there. A published version cannot change, so a changed release needs a new version.", ), ], )