From 0ba59e35e2ccf36fb47c21cb02b719120f32787f Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Wed, 4 Nov 2026 15:25:45 +0100 Subject: [PATCH 1/5] feat(client)!: download releases into the local registry download() now copies a release into the local registry, the first local registry of the list that is not read-only. It resolves the release like open(). Then plan_download() plans the copies, and run_plan() runs them. The requested release and each payload source that the local registry lacks or holds damaged come from the first registry other than the local registry that provides them. So a release and its payload source can come from two registries. If the local registry holds a complete copy of the release, download() copies only the payload sources that the local registry lacks. A client without a writable local registry raises TimeFValidationError. A registry over the directory of the local registry counts as the local registry, also through a symlink or through a case variant of the name on a volume that ignores case. So the local registry is never copied onto itself. download() never builds. For a release that only InstalledConnectors() provides, it raises TimeNetDatasetNotFoundError, which names open() and load(). So download() of "latest" fails when an installed connector declares a newer version than any registry holds. Without a version, download() takes the first source that serves the dataset, as open() does. For an exact version that no source provides, an installed connector that builds another version gives TimeNetVersionUnavailableError, as in open(). A registry that lists the release but does not serve its files gives the same access error as open(). Other errors name the local registry: force=True or a damaged copy when only the local registry holds the release. copy_release() takes force=True, which copies the requested release again into a local registry. A complete payload source stays as it is. Other files still raise TimeNetVersionConflictError. A registry that is not local refuses force=True with TimeFValidationError, because it never writes a published version again. For each release that it copies into a local registry, copy_release() sends BytesMoved events with the release as subject. So a build that copies a parent sends them too. A copy step of a plan passes its force flag to copy_release(). A read of a damaged copy in the local registry ends its TimeNetFileError with a hint that repairs it: "Copy an intact release into the local registry with download('')." when another registry provides the release, or the timenet-build command that builds it again when only the installed connectors provide it. A damaged copy outside the local registry gets no hint. timenet download shows the same output as timenet-build: a status line when the copy of each release starts and ends, a progress bar for each release in a terminal, and "Nothing to do" when the local registry already holds the release and its payload sources. It prints the release directory in the local registry to stdout. With --force, it copies the release again, as download(force=True) does. The help of timenet cache clear now says that --all removes the default local registry with its downloads. The interim download() into TIMENET_STORAGE, _reject_unhosted_access() and BaseRegistry.download_version() are gone. BREAKING CHANGE: download() copies into the local registry, refuses a copy in the local registry with other files, and takes progress= events instead of progress_cb=. BaseRegistry.download_version() is gone. Use TimeNet.download() instead. download() of "latest" reads the newest version of every source in the list, so it needs each registry to be reachable. Part of TNET-157. --- packages/timenet/src/timenet/_fs.py | 40 +- packages/timenet/src/timenet/cache.py | 6 + packages/timenet/src/timenet/cli/app.py | 85 ++-- packages/timenet/src/timenet/cli/ui.py | 14 +- packages/timenet/src/timenet/client.py | 202 ++++++--- .../timenet/src/timenet/coordinator/build.py | 1 + .../timenet/src/timenet/coordinator/plan.py | 97 ++++- .../timenet/src/timenet/coordinator/read.py | 42 +- packages/timenet/src/timenet/dataset/edit.py | 2 +- packages/timenet/src/timenet/progress.py | 8 +- .../timenet/src/timenet/registry/_copy.py | 77 +++- packages/timenet/src/timenet/registry/base.py | 49 +-- .../timenet/src/timenet/registry/local.py | 28 +- .../timenet/tests/coordinator/test_build.py | 5 +- .../tests/reader/test_shared_storage.py | 3 +- packages/timenet/tests/registry/test_copy.py | 70 +++- .../timenet/tests/registry/test_remote.py | 53 ++- packages/timenet/tests/registry/test_s3.py | 9 +- .../timenet/tests/registry/test_version.py | 8 - packages/timenet/tests/test_cli.py | 43 +- packages/timenet/tests/test_client.py | 392 +++++++++++++++--- packages/timenet/tests/test_fs.py | 19 +- 22 files changed, 957 insertions(+), 296 deletions(-) diff --git a/packages/timenet/src/timenet/_fs.py b/packages/timenet/src/timenet/_fs.py index 97c7bf03e..4580bd87a 100644 --- a/packages/timenet/src/timenet/_fs.py +++ b/packages/timenet/src/timenet/_fs.py @@ -1,4 +1,4 @@ -"""Replace a directory with a complete staging directory.""" +"""Compare directories, and replace a directory with a complete staging directory.""" from collections.abc import Generator from contextlib import contextmanager @@ -7,6 +7,44 @@ import uuid +def same_directory(a: Path, b: Path) -> bool: + """Check whether two paths name the same directory, as the file system sees them. + + Two existing directories compare by their identity on disk, so a symlink or a case variant on a volume that + ignores case matches. An existing path never matches a missing one. Two missing paths match if their nearest + existing parents are the same directory and the names below those parents are equal as written. + + Args: + a: The first path, whether or not it exists. + b: The second path, whether or not it exists. + + Returns: + ``True`` if both paths name the same directory. + """ + if a.exists() or b.exists(): + # On a volume that ignores case, a case variant of an existing path exists too. + return a.exists() and b.exists() and a.samefile(b) + return _below_existing_parent(a.resolve()) == _below_existing_parent(b.resolve()) + + +def _below_existing_parent(path: Path) -> tuple[int, int, tuple[str, ...]]: + """Identify a missing path by its nearest existing parent and the names below that parent. + + Args: + path: A missing absolute path without symlinks. + + Returns: + The parent's device and inode numbers, followed by the names below it as written. + """ + for depth, parent in enumerate(path.parents, start=1): + try: + info = parent.stat() + except OSError: + continue + return info.st_dev, info.st_ino, path.parts[len(path.parts) - depth :] + return 0, 0, path.parts # nothing exists, not even the root + + def publish_directory(staging: Path, final: Path) -> None: """Rename ``staging`` to ``final``, replacing any existing directory. diff --git a/packages/timenet/src/timenet/cache.py b/packages/timenet/src/timenet/cache.py index f8c6e5e5a..83f6c9406 100644 --- a/packages/timenet/src/timenet/cache.py +++ b/packages/timenet/src/timenet/cache.py @@ -4,6 +4,10 @@ download cache. All three sit under the home directory. """ +# Glue: the "download storage" of this module is the storage directory, settings().storage_dir. +# download() copies into the local registry now, so the storage directory holds only the copies that remote +# registries cache. A later change replaces this module and its view of the storage directory. + from dataclasses import dataclass import os from pathlib import Path @@ -49,6 +53,7 @@ def cached_datasets() -> list[CachedDataset]: The cached datasets, sorted by (location, id, version). """ config = settings() + # Glue: the storage directory, as the note at the top of this module says. found = _scan(config.registry_path, "registry") + _scan(config.storage_dir, "storage") return sorted(found, key=lambda dataset: (dataset.location, dataset.dataset_id, dataset.version)) @@ -78,6 +83,7 @@ def clear_cache(*, include_registry: bool) -> tuple[int, list[Path]]: reporting space it never freed. """ config = settings() + # Glue: the storage directory, as the note at the top of this module says. targets = [config.storage_dir, config.cache_dir] if include_registry: targets.append(config.registry_path) diff --git a/packages/timenet/src/timenet/cli/app.py b/packages/timenet/src/timenet/cli/app.py index 4b27f4610..209c42639 100644 --- a/packages/timenet/src/timenet/cli/app.py +++ b/packages/timenet/src/timenet/cli/app.py @@ -1,21 +1,21 @@ """The consumer command-line interface, the command-line mirror of the :class:`~timenet.client.TimeNet` SDK.""" from collections.abc import Callable, Iterable -from pathlib import Path -import sys +from functools import partial from typing import TypeVar from rich.console import Console -from rich.progress import BarColumn, DownloadColumn, Progress, TextColumn, TimeRemainingColumn, TransferSpeedColumn from rich.table import Table import typer from timenet.cache import CachedDataset, cached_datasets, clear_cache, human_bytes, raw_cache_size from timenet.cli.runner import run_cli -from timenet.cli.ui import console as ui +from timenet.cli.ui import console as ui, progress_renderer from timenet.client import TimeNet, default_sources +from timenet.progress import ProgressCallback, ProgressEvent, StepStarted from timenet.types import DatasetMetadata, Domain, License, TaskType from timenet.types.access import access_requirements, effective_access +from timenet.types.references import split_ref from timenet.types.tasks import TASKS @@ -28,6 +28,13 @@ "-r", help="Registry URL or path that this command reads (else $TIMENET_REGISTRY, then timenet://).", ) +_download_registry_option = typer.Option( + None, + "--registry", + "-r", + help="Registry URL or path to put after the local registry in the list of sources. The release resolves " + "through the whole list.", +) T = TypeVar("T") @@ -184,49 +191,41 @@ def info(dataset_id: str, registry: str | None = _registry_option) -> None: @app.command() -def download(dataset_id: str, registry: str | None = _registry_option) -> None: - """Download a dataset version and print its local directory to stdout. +def download( + dataset_id: str, + registry: str | None = _download_registry_option, + force: bool = typer.Option( + False, "--force", help="Copy the release again, even if the local registry holds a complete copy." + ), +) -> None: + """Copy a release and its payload sources into the local registry. - Files go to ``$TIMENET_STORAGE`` or ``/storage``. + Print the release directory to standard output. """ - client = _client(registry) - manifest = client.get(dataset_id) - files = list(manifest.files.all_files()) - total = sum(part.size for part in files) - ui.status("⬇️", f"Downloading '{dataset_id}' ({len(files)} files, {human_bytes(total)})…") - target = _download(client, dataset_id, total) - ui.success(f"Downloaded '{dataset_id}' ({target.name})") + sources = default_sources(remote_registry=registry) + started: list[StepStarted] = [] + with progress_renderer() as show: + target = TimeNet(sources=sources).download( + dataset_id, force=force, progress=partial(_keep_steps, started, show) + ) + if not started: + # The local registry already held the release and its payload sources. + ui.success(f"Nothing to do: '{split_ref(dataset_id)[0]}' {target.name} is already in {sources[0]}") typer.echo(str(target)) -def _download(client: TimeNet, dataset_id: str, total: int) -> Path: - """Download a version, rendering a live progress bar on stderr when the terminal supports it. +def _keep_steps(started: list[StepStarted], show: ProgressCallback | None, event: ProgressEvent) -> None: + """Keep each step event in ``started``, and pass every event on to ``show``. Args: - client: The client to download through. - dataset_id: The dataset id, optionally followed by ``@version``. - total: The total file size in bytes for the progress bar. - - Returns: - The local version directory. + started: Receives the step events. + show: The callback that shows the events, or ``None`` for a quiet console. + event: The progress event. """ - if ui.quiet or not sys.stderr.isatty(): - return client.download(dataset_id) - progress = Progress( - TextColumn("[bold blue]{task.description}"), - BarColumn(bar_width=None), - "[progress.percentage]{task.percentage:>3.0f}%", - "•", - DownloadColumn(), - "•", - TransferSpeedColumn(), - "•", - TimeRemainingColumn(), - console=ui.rich, - ) - with progress: - task = progress.add_task("downloading", total=total) - return client.download(dataset_id, progress_cb=lambda n: progress.advance(task, n)) + if isinstance(event, StepStarted): + started.append(event) + if show is not None: + show(event) cache_app = typer.Typer(help="Inspect and clear the local cache.", no_args_is_help=True) @@ -270,14 +269,14 @@ def _cache_table(datasets: list[CachedDataset]) -> Table: @cache_app.command("clear") def cache_clear( - all_: bool = typer.Option(False, "--all", help="Also remove the local registry (built datasets)."), + all_: bool = typer.Option(False, "--all", help="Also remove the default local registry (builds and downloads)."), yes: bool = typer.Option(False, "--yes", "-y", help="Skip the confirmation prompt."), ) -> None: - """Delete cached data. Clears downloads by default. The ``--all`` flag also removes built datasets.""" + """Delete cached data: the storage directory and the raw cache. ``--all`` also removes the default local registry.""" if not yes: - what = "downloaded datasets and the raw cache" + what = "the storage directory and the raw cache" if all_: - what += " and the local registry (built datasets)" + what += " and the local registry (builds and downloads)" typer.confirm(f"Remove {what}?", abort=True) freed, removed = clear_cache(include_registry=all_) typer.echo(f"Freed {human_bytes(freed)} from {len(removed)} location(s).") diff --git a/packages/timenet/src/timenet/cli/ui.py b/packages/timenet/src/timenet/cli/ui.py index f2c8b0ffa..b4ee73d54 100644 --- a/packages/timenet/src/timenet/cli/ui.py +++ b/packages/timenet/src/timenet/cli/ui.py @@ -125,7 +125,7 @@ def _download_bars() -> Progress: def _bar_reporter(progress: Progress) -> Callable[[BytesMoved], None]: - """Create a callback that displays a progress bar for each transfer URL. + """Create a callback that displays a progress bar for each transfer. Args: progress: The display that holds the bars. @@ -136,10 +136,10 @@ def _bar_reporter(progress: Progress) -> Callable[[BytesMoved], None]: tasks: dict[str, TaskID] = {} def report(event: BytesMoved) -> None: - task_id = tasks.get(event.url) + task_id = tasks.get(event.subject) if task_id is None: - task_id = progress.add_task("download", name=event.url.rsplit("/", 1)[-1], total=event.total) - tasks[event.url] = task_id + task_id = progress.add_task("download", name=event.subject.rsplit("/", 1)[-1], total=event.total) + tasks[event.subject] = task_id progress.update(task_id, completed=event.done, total=event.total) return report @@ -161,10 +161,10 @@ def _text_download_reporter() -> Callable[[BytesMoved], None]: def report(event: BytesMoved) -> None: # ~20% steps when the size is known, else one line every 50 MB. step = event.done * 5 // event.total if event.total else event.done // _UNKNOWN_TOTAL_STEP_BYTES - if last.get(event.url) == step: + if last.get(event.subject) == step: return - last[event.url] = step - name = event.url.rsplit("/", 1)[-1] + last[event.subject] = step + name = event.subject.rsplit("/", 1)[-1] if event.total: console.status("⬇️", f"{name} {event.done * 100 // event.total}% ({human_bytes(event.total)})") else: diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index 0a2a49918..265a00155 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -1,4 +1,14 @@ -"""Browse and load datasets with the :class:`TimeNet` client.""" +"""The :class:`TimeNet` class is the SDK's single entry point for using TimeNet from code. + +A client reads releases through an ordered list of sources: registries and +``InstalledConnectors()`` (see :mod:`timenet.sources`). Its methods ``open``, ``load``, and +``open_torch`` read a release where it is, through the read coordinator in +:mod:`timenet.coordinator.read`. If only ``InstalledConnectors()`` provides a release, they build it +into the first :class:`~timenet.registry.LocalRegistry` of the list that is not +``read_only=True``. The method ``download`` copies a release into that local registry, from the first +registry of the list that provides it. The methods ``list``, ``get``, and ``search`` ask one registry of the +list. :func:`default_sources` makes the list of ``TimeNet()``. +""" # The public API has a method named ``list``. Deferred annotations keep the type hint # ``list[str]`` resolved to the builtin type, not to the method. @@ -7,25 +17,41 @@ from collections.abc import Sequence import os from pathlib import Path +import shlex from typing import TYPE_CHECKING, TypeAlias, TypeVar +from timenet.builders import BuildOptions from timenet.config import settings -from timenet.coordinator.read import read_release +from timenet.coordinator.build import run_plan +from timenet.coordinator.plan import Lookups, plan_download +from timenet.coordinator.read import raise_not_found, read_release from timenet.dataset import TimeFDataset from timenet.errors import ( TimeFValidationError, TimeNetAccessError, TimeNetDatasetNotFoundError, + TimeNetError, + TimeNetFileError, TimeNetMissingParentError, + TimeNetVersionUnavailableError, ) from timenet.manifest import Manifest +from timenet.progress import ProgressCallback from timenet.reader import LoadPlan, TimeFReader from timenet.registry import BaseRegistry, LocalRegistry, open_registry from timenet.registry._paths import release_dir -from timenet.registry.base import ProgressCallback -from timenet.sources import InstalledConnectors, ReleaseSource +from timenet.registry.local import is_same_registry +from timenet.sources import ( + Found, + FoundInConnector, + FoundInRegistry, + InstalledConnectors, + NotFound, + NotServed, + ReleaseSource, + resolve_release, +) from timenet.types import DatasetMetadata, Domain, License, Task -from timenet.types.access import require_open_access from timenet.types.references import DatasetRef, split_ref @@ -221,35 +247,67 @@ def download( version: str | None = None, *, force: bool = False, - progress_cb: ProgressCallback | None = None, + progress: ProgressCallback | None = None, ) -> Path: - """Download a dataset version and the data needed to read it. + """Copy a release and its payload sources into the store, from the first source that provides it. - The client uses the first registry other than the local registry, or the local registry if it is the only one. - Files go to ``$TIMENET_STORAGE`` or ``/storage``. + The release resolves as in :meth:`open`, without a build. The release and each payload source that the + local registry lacks or holds damaged come from the first registry other than the local registry that + provides them. Args: dataset_id: The dataset ID. - version: The version string, or ``None`` for the latest. - force: Download again, even if an up-to-date copy already exists. - progress_cb: A callback that receives each file's byte count after download. + version: An exact version, ``"latest"``, or ``None``, as in :meth:`open`. + force: Copy the release again, even if the local registry holds a complete copy. + progress: Receives the progress events of the copy: a step for each release that it copies + or repairs, and the bytes of each such release as they arrive. Returns: - The local ``///`` directory. - """ - # TODO(adrian): Remove temporary glue which maps download() onto a copy from one registry - # into the storage directory, as before the list of sources, until a download into the - # local registry replaces it. + The release directory in the local registry. + + Raises: + TimeFValidationError: If the sources lack a writable local registry, or both arguments specify a version. + TimeNetAccessError: If a required release is only listed or a remote release is not open. + TimeNetDatasetNotFoundError: If no registry other than the local registry provides the release. + TimeNetVersionUnavailableError: If the version is missing and its connector builds another version. + TimeNetMissingParentError: If no registry provides a payload source that the store lacks. + TimeNetVersionConflictError: If the local registry holds a release of the copy with different files. + TimeNetFileError: If a copied file is damaged or no registry can replace a damaged local copy. + TimeFFormatError: If a payload source differs from its recorded files or the payload sources form a cycle. + """ # ruff: ignore[docstring-extraneous-exception] (raised by the plan and the copies) dataset_id, version = _resolve_ref(dataset_id, version) - registry = self._catalog() - self._reject_unhosted_access(dataset_id, version) - manifest = registry.get_manifest(dataset_id, version) - resolved = str(manifest.metadata.dataset_version) - target = release_dir(settings().storage_dir, DatasetRef.parse(f"{dataset_id}@{resolved}")) - # The registry owns the fetch: the base implementation stages each file and swaps atomically, - # and a remote registry overrides it to resolve and stream every file in parallel. - registry.download_version(dataset_id, resolved, target, force=force, manifest=manifest, progress_cb=progress_cb) - return target + store = self._store + if store is None: + # Without a local registry, a download has no place to go. + raise TimeFValidationError( + "download() needs a writable local registry to copy into. " + "Add a LocalRegistry without read_only=True to the sources" + ) + lookups = Lookups() + answer = lookups.resolve(self._sources, dataset_id, version) + if isinstance(answer, NotServed): + # A registry lists the release, but no source serves its files. + error = answer.access_error(dataset_id) + error.hint = _access_hint(dataset_id, self._sources) + raise error + if isinstance(answer, NotFound): + # No source provides the release. + try: + raise_not_found(dataset_id, version, self._sources) + except TimeNetVersionUnavailableError: + # An installed connector builds another version, and the error names it. + raise + except TimeNetDatasetNotFoundError as error: + raise TimeNetDatasetNotFoundError(f"{error} Add a registry that holds it to the sources.") from error + release = DatasetRef(dataset_id=dataset_id, version=answer.version) + try: + plan = plan_download(release, store, sources=self._sources, force=force, lookups=lookups) + except TimeNetFileError as error: + # The local registry holds a damaged copy that no other registry provides. + error.hint = self._file_hint(error) + raise + run_plan(plan, store, options=BuildOptions(progress=progress)) + return release_dir(store.root, release) def open(self, dataset_id: str, version: str | None = None) -> TimeFReader: """Open a dataset version for lazy reads, where it is. @@ -295,7 +353,9 @@ def open(self, dataset_id: str, version: str | None = None) -> TimeFReader: TimeNetMissingParentError: If no source provides a payload source, or a release that a build needs. The message says which source to add. TimeNetFileError: If a release that the call opens in a local registry has a missing - file or a file of another size. + file or a file of another size. If the damaged copy is in the local registry, the message says + how to replace it: with the :meth:`download` that copies an intact release from a + registry, else with a new build, if an installed connector builds the release. TimeFFormatError: If a payload source differs from the files that the version pins, for example because its build does not reproduce them. TimeNetBuildError: If a build runs but fails. @@ -306,6 +366,9 @@ def open(self, dataset_id: str, version: str | None = None) -> TimeFReader: except TimeNetAccessError as error: error.hint = _access_hint(dataset_id, self._sources) raise + except TimeNetFileError as error: + error.hint = self._file_hint(error) + raise except TimeNetMissingParentError as error: error.hint = _not_provided_hint(self._sources) raise @@ -348,7 +411,7 @@ def load(self, dataset_id: str, version: str | None = None) -> TimeFDataset: return reader.read(load=LoadPlan.ALL_EAGER) def _catalog(self) -> BaseRegistry: - """Select a registry from the configured sources. + """Return the registry that :meth:`list`, :meth:`get`, and :meth:`search` ask. Returns: The first registry other than the local registry, or the local registry if it is the only one. @@ -362,30 +425,35 @@ def _catalog(self) -> BaseRegistry: registries = [source for source in self._sources if isinstance(source, BaseRegistry)] registry = next((registry for registry in registries if registry is not self._store), self._store) if registry is None: - raise TimeFValidationError("list(), get(), search() and download() need a registry in the sources") + raise TimeFValidationError("list(), get() and search() need a registry in the sources") return registry - # TODO(adrian): Remove temporary glue which checks access only for the interim download() - # until a later change replaces download(). - def _reject_unhosted_access(self, dataset_id: str, version: str | None) -> None: - """Reject datasets without open access when the registry requires it. + def _file_hint(self, error: TimeNetFileError) -> str | None: + """Tell the caller how to replace the damaged copy that an error names, if the store holds it. + + :meth:`download` replaces only a copy in the store. So a damaged copy in another local + registry gets no hint. Args: - dataset_id: The dataset ID without a version suffix. - version: The resolved version, or ``None`` for the latest. + error: The error that names the damaged release. - Raises: - TimeNetAccessError: If the dataset lacks open access and the registry requires it. + Returns: + A repair hint, or ``None`` if the local registry has no damaged copy or the lookup raises a TimeNet error. """ - registry = self._catalog() - if not registry.requires_open_access: - return + store = self._store + if store is None: + return None + damaged = error.release try: - manifest = registry.get_manifest(dataset_id, version) - require_open_access(manifest.metadata, manifest.lineage) - except TimeNetAccessError as error: - error.hint = _unhosted_hint(dataset_id) - raise + held = store.find_manifest(damaged) + if held is None or store.damage(damaged, held) is None: + # The local registry holds no damaged copy, so the error is about another registry. + return None + found = _copy_source(self._sources, store, damaged) + except TimeNetError: + # The hint must not hide the error that it explains, for example if a registry is offline. + return None + return _repair_hint(damaged, found, store) def open_torch( self, dataset_id: str, version: str | None = None, *, load: LoadPlan = LoadPlan.ALL_EAGER @@ -445,12 +513,44 @@ def _access_hint(dataset_id: str, sources: Sequence[ReleaseSource]) -> str: return f"No source serves its files. Add InstalledConnectors() to the sources, or build it locally with {build}." -# TODO(adrian): Remove temporary glue which gives a hint only for the interim download() until a -# later change replaces download(). -def _unhosted_hint(dataset_id: str) -> str: - """Suggest a local build for a dataset whose registry does not serve its files. +def _copy_source(sources: Sequence[ReleaseSource], store: LocalRegistry, release: DatasetRef) -> Found | None: + """Return the answer for a release of the first source other than the store that provides it. + + :meth:`TimeNet.download` copies the release only if a registry other than the store gave this answer. + + Args: + sources: The sources of the client. + store: The local registry. + release: The exact release. + + Returns: + The answer of the source that serves or builds the release, or ``None`` if no source other + than the store provides it. + """ + others = [source for source in sources if not is_same_registry(source, store)] + answer = resolve_release(others, release.dataset_id, str(release.version)) + return answer if isinstance(answer, Found) else None + + +def _repair_hint(release: DatasetRef, found: Found | None, store: LocalRegistry) -> str: + """Tell the caller how to replace a damaged copy of a release in the store. + + Args: + release: The release. + found: The answer that :func:`_copy_source` returns for the release. + store: The local registry. Returns: - A hint with the build command. + The hint as a full sentence. It names :meth:`TimeNet.download` if a registry other than the + store provides the release, else the build that replaces the copy, if an installed connector + builds the release. """ - return f"TimeNet does not host its data, so build it locally with `timenet-build build {dataset_id}`." + if isinstance(found, FoundInRegistry) and not is_same_registry(found.registry, store): + # A registry other than the local registry provides the release. + return f"Copy an intact release into the local registry with download('{release}')." + if isinstance(found, FoundInConnector): + # Only an installed connector provides the release. + command = f"timenet-build build {release.dataset_id} --out {shlex.quote(str(store.root))} --force" + return f"Build it again with `{command}`." + # Only the local registry holds the release. + return "No registry other than the local registry provides it, so download() cannot replace it." diff --git a/packages/timenet/src/timenet/coordinator/build.py b/packages/timenet/src/timenet/coordinator/build.py index 6ac98f8e4..d13243ef1 100644 --- a/packages/timenet/src/timenet/coordinator/build.py +++ b/packages/timenet/src/timenet/coordinator/build.py @@ -83,6 +83,7 @@ def run_plan(plan: Plan, store: LocalRegistry, *, options: BuildOptions | None = step.registry, store, with_payload_sources=step.with_payload_sources, + force=step.force, progress=options.progress, ) case BuildStep(): diff --git a/packages/timenet/src/timenet/coordinator/plan.py b/packages/timenet/src/timenet/coordinator/plan.py index 219a5635f..bc407256d 100644 --- a/packages/timenet/src/timenet/coordinator/plan.py +++ b/packages/timenet/src/timenet/coordinator/plan.py @@ -22,7 +22,7 @@ find_payload_sources, validate_manifest_reference, ) -from timenet.registry.local import LocalRegistry +from timenet.registry.local import LocalRegistry, is_same_registry from timenet.registry.version import DatasetVersion from timenet.sources import ( Found, @@ -64,6 +64,8 @@ class CopyStep: """The releases with steps that must finish before this step runs.""" with_payload_sources: bool = True """Also copy the payload sources that the manifest of the release lists.""" + force: bool = False + """Copy the release again, also if the local registry holds it.""" action: ClassVar[Literal["copy"]] = "copy" """The kind of step, for display.""" @@ -204,6 +206,8 @@ class _Graph: """The sources in preference order. A read reads the output like any other source.""" open_only: bool """True if each release that the walk visits must declare ``access: open``.""" + builds: bool = True + """True if the plan may build a release that only installed connectors provide.""" releases: dict[DatasetRef, _Release] = field(default_factory=dict) """The nodes, in the order in which the walk found them.""" pins: dict[DatasetRef, dict[DatasetRef, PayloadSource]] = field(default_factory=dict) @@ -217,9 +221,9 @@ class _Graph: def build_sources(self) -> list[ReleaseSource]: """The sources without the output, which is never a source of its own copies and builds.""" output = self.output - return [source for source in self.sources if output is None or not _is_same_registry(source, output)] + return [source for source in self.sources if output is None or not is_same_registry(source, output)] - def visit( + def visit( # ruff: ignore[too-many-arguments] self, reference: DatasetRef, needed_by: DatasetRef | None, @@ -227,6 +231,7 @@ def visit( *, needed_in_local_registry: bool, answer: ReleaseLookup | None = None, + force: bool = False, ) -> None: """Add a release and its dependencies to the graph. @@ -234,11 +239,12 @@ def visit( copy in the output instead of a read. Args: - reference: The exact release to add. - needed_by: The release that needs ``reference``, or ``None`` for the requested release. - pinned: File requirements from ``needed_by``, or ``None`` for a parent dataset. + reference: The release to add. + needed_by: The release that requires ``reference``, or ``None`` for the requested release. + pinned: The file records that ``needed_by`` requires, or ``None`` for a parent dataset. needed_in_local_registry: True if the plan needs a complete copy in the output. answer: The answer of the sources for ``reference``, or ``None`` to ask them. + force: Copy the release again, even if the output holds a complete copy. """ role: Role = "dataset" if pinned is None else "payload-source" if pinned is not None and needed_by is not None: @@ -250,7 +256,7 @@ def visit( if (role, needed_in_local_registry) == (found.role, found.needed_in_local_registry): return if needed_in_local_registry: - self.local(reference, needed_by, role) + self.local(reference, needed_by, role, force=force) else: self.read(reference, needed_by, role, answer) @@ -302,16 +308,19 @@ def read( else: raise self.missing(reference, needed_by, role, self.sources) - def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role) -> None: + def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role, *, force: bool = False) -> None: """Plan a complete copy of a release in the output, from the output itself, a copy or a build. Args: reference: The exact release. needed_by: The release that needs ``reference``, or ``None`` for the requested release. role: Whether the release is a dataset or a payload source. + force: Copy the release again, even if the output holds a complete copy. Raises: TimeFValidationError: If the plan has no output. + TimeNetDatasetNotFoundError: If the plan builds nothing and only installed connectors provide the release, + no source provides the requested release, or ``force`` is set and only the output holds it. TimeNetFileError: If the output holds a damaged copy that no other registry provides. TimeNetAccessError: If a source lists the release but no source serves it. TimeNetMissingParentError: If no source provides the release. @@ -320,7 +329,7 @@ def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role) if output is None: raise TimeFValidationError(f"{reference} must be in a local registry, but the read has none") held, damage = self.in_output(reference) - if held is not None: + if held is not None and not force: if self.open_only: require_open_access(held.metadata, held.lineage) sources = list(held.payload_sources.values()) if role == "dataset" else [] @@ -330,11 +339,20 @@ def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role) return answer = self.lookups.resolve(self.build_sources, reference.dataset_id, str(reference.version)) if isinstance(answer, FoundInRegistry): - self.copy(reference, role, answer.registry) + self.copy(reference, role, answer.registry, force=force) elif damage is not None: # A copy from another registry repairs a damaged copy. Nothing else replaces it. raise damage + elif held is not None: + # With force, the copy needs a registry other than the output, and none provides the release. + raise TimeNetDatasetNotFoundError( + f"No registry other than the local registry provides {reference}, so download(force=True) cannot copy " + "it again." + ) elif isinstance(answer, FoundInConnector): + if not self.builds: + # A download builds nothing. + raise TimeNetDatasetNotFoundError(_not_built(reference)) self.build(reference, role, answer.builder) elif isinstance(answer, NotServed): raise answer.access_error(reference.dataset_id) @@ -342,7 +360,7 @@ def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role) # No source provides the release, or only the output does. raise self.missing(reference, needed_by, role, (output, *self.build_sources)) - def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> None: + def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry, *, force: bool = False) -> None: """Add a copy step and plan the payload sources that the copy does not bring. A dataset copy brings each payload source that ``registry`` serves. Each other payload source that the output @@ -352,6 +370,7 @@ def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> Non reference: The exact release to copy. role: Whether the release is a dataset or a payload source. registry: The registry to copy from. + force: Copy the release again, also if the output holds a complete copy. Raises: TimeNetDatasetNotFoundError: If the release is absent from the registry. @@ -374,7 +393,7 @@ def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> Non if self.open_only: for found in (*sources.fetched.values(), *sources.held.values()): require_open_access(found.metadata, found.lineage) - step = CopyStep(reference, registry, with_payload_sources=role == "dataset") + step = CopyStep(reference, registry, with_payload_sources=role == "dataset", force=force) # The node holds its final dependencies before the walk visits them. So a release of the walk # that depends on it closes a cycle and does not look it up again. self.releases[reference] = _Release(role, True, step, tuple(sources.missing)) @@ -616,19 +635,57 @@ def plan_steps( return root, graph.create_ordered_steps() -def _is_same_registry(source: ReleaseSource, output: LocalRegistry) -> bool: - """Check whether a source is the output registry. +def plan_download( + release: DatasetRef, + store: LocalRegistry, + *, + sources: Sequence[ReleaseSource], + force: bool = False, + lookups: Lookups | None = None, +) -> Plan: + """Plan the copies that put a release and its payload sources into the local registry. + + Each release that the local registry lacks comes from the first registry other than the local registry + that provides it. The plan builds nothing. + + Args: + release: The exact release. + store: The store to copy into. + sources: The sources, in order. The store, and each registry over its directory, do not count. + force: Copy the release again, even if the local registry holds a complete copy. + lookups: The answers and manifests that resolved ``release``, or ``None`` to ask the sources. + + Returns: + The copy steps, in the order in which they run. + + Raises: + TimeNetDatasetNotFoundError: If only installed connectors provide a release that the store + lacks, or if ``force`` is set and only the local registry holds the release. + TimeNetFileError: If the local registry holds a damaged copy that no other registry provides. + TimeNetAccessError: If a source lists a release that the local registry lacks but no source serves it. + TimeNetMissingParentError: If no registry provides a payload source that the local registry lacks. + TimeFFormatError: If a payload source differs from the files that a release pins. + """ # ruff: ignore[docstring-extraneous-exception] - raised by the walk and as values + others = [source for source in sources if not is_same_registry(source, store)] + graph = _Graph(store, others, open_only=False, builds=False, lookups=lookups or Lookups()) + graph.visit(release, None, needed_in_local_registry=True, force=force) + graph.check() + return Plan(tuple(graph.create_ordered_steps())) + + +def _not_built(release: DatasetRef) -> str: + """Describe a release that only installed connectors provide, in a plan that builds nothing. Args: - source: The source to compare. - output: The destination registry. + release: The release. Returns: - ``True`` for the same object or local registries with the same resolved directory. + The error message. """ - if source is output: - return True - return isinstance(source, LocalRegistry) and source.root.resolve() == output.root.resolve() + return ( + f"{release} is not built yet: only an installed connector provides it. open() or load() builds it into " + "the local registry." + ) def describe_searched_sources(sources: Iterable[ReleaseSource], reference: DatasetRef) -> str: diff --git a/packages/timenet/src/timenet/coordinator/read.py b/packages/timenet/src/timenet/coordinator/read.py index a1a771529..287ff7f5d 100644 --- a/packages/timenet/src/timenet/coordinator/read.py +++ b/packages/timenet/src/timenet/coordinator/read.py @@ -5,7 +5,7 @@ """ from collections.abc import Sequence -from typing import cast +from typing import NoReturn, cast from timenet.builders import declared, find_card from timenet.coordinator.build import run_plan @@ -86,6 +86,31 @@ def read_release( ) +def raise_not_found(dataset_id: str, version: str | None, sources: Sequence[ReleaseSource]) -> NoReturn: + """Raise the error for a release that no source provides. + + For an exact version, an installed connector that builds another version gives a more exact error. + + Args: + dataset_id: The dataset ID. + version: An exact version, ``"latest"``, or ``None`` or ``""`` for no version. + sources: The sources searched for the release. + + Raises: + TimeNetVersionUnavailableError: If the installed connector declares another version than the exact request. + TimeNetInvalidCardError: If the card of the connector cannot be read. + TimeNetDatasetNotFoundError: Otherwise, with a message that names the searched sources. + """ # ruff: ignore[docstring-extraneous-exception] - raised by declared + requested = None if is_latest(version) else DatasetRef.parse(f"{dataset_id}@{version}") + connectors = any(isinstance(source, InstalledConnectors) for source in sources) + if requested is not None and connectors and find_card(dataset_id) is not None: + # The installed connector builds another version. + declared(dataset_id, requested.version) + names = ", ".join(str(source) for source in sources) + wanted = dataset_id if requested is None else str(requested) + raise TimeNetDatasetNotFoundError(f"{wanted} is in none of the sources: {names}.") + + def _walk( dataset_id: str, version: str | None, *, sources: Sequence[ReleaseSource], store: LocalRegistry | None ) -> tuple[_Graph, DatasetRef]: @@ -96,23 +121,16 @@ def _walk( Raises: TimeNetDatasetNotFoundError: If no source provides the release. - """ + """ # ruff: ignore[docstring-extraneous-exception] - raised by raise_not_found graph = _Graph(store, sources, open_only=False) answer = graph.lookups.resolve(sources, dataset_id, version) if isinstance(answer, NotFound): - requested = None if is_latest(version) else DatasetRef.parse(f"{dataset_id}@{version}") - if requested is not None: + if not is_latest(version): # Nothing but a registry replaces a damaged copy in the local registry. - damage = graph.in_output(requested)[1] + damage = graph.in_output(DatasetRef.parse(f"{dataset_id}@{version}"))[1] if damage is not None: raise damage - connectors = any(isinstance(source, InstalledConnectors) for source in sources) - if connectors and find_card(dataset_id) is not None: - # The installed connector builds another version. - declared(dataset_id, requested.version) - names = ", ".join(str(source) for source in sources) - wanted = dataset_id if requested is None else str(requested) - raise TimeNetDatasetNotFoundError(f"{wanted} is in none of the sources: {names}.") + raise_not_found(dataset_id, version, sources) release = DatasetRef(dataset_id=dataset_id, version=answer.version) # The walk takes the answer, so that a read without an exact version does not ask again for the exact version. graph.visit(release, None, needed_in_local_registry=False, answer=answer) diff --git a/packages/timenet/src/timenet/dataset/edit.py b/packages/timenet/src/timenet/dataset/edit.py index 2275fc5e2..c51fba7eb 100644 --- a/packages/timenet/src/timenet/dataset/edit.py +++ b/packages/timenet/src/timenet/dataset/edit.py @@ -509,7 +509,7 @@ def edit_version( The derived version reuses the base's payload files, so it is written into the base's registry as ``///``, where readers find the base. To derive a version in - another registry, copy the base there first, for example with ``download_version``, then edit + another registry, copy the base there first, for example with ``copy_release()``, then edit the copy. The derived version keeps the base's parent cards, lineage, and import map, without the diff --git a/packages/timenet/src/timenet/progress.py b/packages/timenet/src/timenet/progress.py index 87fb8a9b6..5ce2a2593 100644 --- a/packages/timenet/src/timenet/progress.py +++ b/packages/timenet/src/timenet/progress.py @@ -42,12 +42,12 @@ class BytesMoved: """The byte count of a transfer in progress. Attributes: - url: The source or destination URL. - done: The number of bytes moved so far. - total: The transfer size in bytes, or ``None`` if unknown. + subject: What moves: a source or target URL, or a copied release ID such as ``"org/name@1.0.0"``. + done: The bytes moved so far. + total: The total byte count, or ``None`` if unknown. """ - url: str + subject: str done: int total: int | None diff --git a/packages/timenet/src/timenet/registry/_copy.py b/packages/timenet/src/timenet/registry/_copy.py index f29131ebb..6b2c71f9c 100644 --- a/packages/timenet/src/timenet/registry/_copy.py +++ b/packages/timenet/src/timenet/registry/_copy.py @@ -4,12 +4,13 @@ Manifests must match for existing releases, except for their build environments. """ +from collections.abc import Callable from pathlib import Path import tempfile -from timenet.errors import TimeNetMissingParentError, TimeNetVersionConflictError +from timenet.errors import TimeFValidationError, TimeNetMissingParentError, TimeNetVersionConflictError from timenet.manifest import Manifest, same_release -from timenet.progress import ProgressCallback, StepFinished, StepStarted +from timenet.progress import BytesMoved, 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 @@ -19,12 +20,13 @@ from timenet.types.references import DatasetRef -def copy_release( +def copy_release( # ruff: ignore[too-many-arguments] reference: DatasetRef, source: BaseRegistry, dest: WritableRegistry, *, with_payload_sources: bool = True, + force: bool = False, progress: ProgressCallback | None = None, ) -> None: """Copy a release and its required payload sources to another registry. @@ -37,21 +39,36 @@ def copy_release( 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 release. + force: Fetch ``reference`` again, even if a local ``dest`` holds a complete copy of it. + progress: Receives :class:`~timenet.progress.StepStarted` and + :class:`~timenet.progress.StepFinished` for each release that is copied or repaired. For + each release that a local ``dest`` gets, it also receives + :class:`~timenet.progress.BytesMoved` events between the two, as the files arrive. Their + ``url`` is the release, and their ``total`` is the sum of the file sizes in its manifest. Raises: TimeNetRegistryError: If ``dest`` is read-only. - TimeNetVersionConflictError: If ``dest`` holds other files or another card for a release being copied. - TimeNetAccessError: If a release fails an open access requirement of ``dest`` or ``source``. + TimeFValidationError: If ``force`` is set and ``dest`` is not local. + TimeNetVersionConflictError: If ``dest`` holds the release, or a payload source of it, with + other files or another card. + TimeNetAccessError: If a release of the copy is not open, and ``dest`` requires open access + or ``source`` requires it for a release that the copy fetches. TimeNetDatasetNotFoundError: If ``source`` does not hold the release. TimeNetMissingParentError: If ``source`` lacks a payload source and ``dest`` holds no complete copy of it. TimeFFormatError: If a manifest names a different release or a pin fails. """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers dest.check_writable(f"a copy of {reference}") + if force and not isinstance(dest, LocalRegistry): + raise TimeFValidationError(f"force=True needs a local dest: {dest} never writes a published version again") 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)] + # source that only dest holds is complete there, so it is never pending. _needs_copy() runs first, + # so that other files raise also with force. + pending = [ + ref + for ref, manifest in manifests.items() + if _needs_copy(reference, ref, manifest, source, dest) or (force and ref == reference) + ] _check_access(manifests, set(pending), source, dest) if not pending: return @@ -61,13 +78,21 @@ def copy_release( for ref in pending: if progress is not None: progress(StepStarted(ref, "copy")) - _copy_one(ref, manifests[ref], source, dest, Path(staging)) + again = force and ref == reference + _copy_one(ref, manifests[ref], source, dest, Path(staging), force=again, progress=progress) if progress is not None: progress(StepFinished(ref, "copy")) -def _copy_one( - reference: DatasetRef, manifest: Manifest, source: BaseRegistry, dest: WritableRegistry, staging: Path +def _copy_one( # ruff: ignore[too-many-arguments] + reference: DatasetRef, + manifest: Manifest, + source: BaseRegistry, + dest: WritableRegistry, + staging: Path, + *, + force: bool, + progress: ProgressCallback | None, ) -> None: """Copy one release without its payload sources to ``dest``, or repair its local copy. @@ -76,11 +101,15 @@ def _copy_one( 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. + staging: A temporary directory for a download from a non-local ``source``. + force: Fetch the release again, even if a local ``dest`` holds a complete copy. + progress: Receives the bytes of a local copy as they arrive. """ # 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) + report = None if progress is None else _bytes_of(reference, manifest, progress) + directory = release_dir(dest.local_root, reference) + source.download_release(reference, directory, force=force, manifest=manifest, progress_cb=report) return if isinstance(source, LocalRegistry): directory = release_dir(source.root, reference) @@ -93,6 +122,28 @@ def _copy_one( dest.put_release(directory) +def _bytes_of(reference: DatasetRef, manifest: Manifest, progress: ProgressCallback) -> Callable[[int], None]: + """Turn the byte counts of the download of one release into :class:`~timenet.progress.BytesMoved` events. + + Args: + reference: The release. + manifest: The manifest of the release, whose file sizes give the total. + progress: Receives the events. + + Returns: + A function that takes the bytes of each file or chunk as it arrives, and sends the bytes so far. + """ + total = sum(part.size for part in manifest.files.all_files()) + done = 0 + + def report(count: int) -> None: + nonlocal done + done += count + progress(BytesMoved(str(reference), done, total)) + + return report + + def _check_access( manifests: dict[DatasetRef, Manifest], fetched: set[DatasetRef], source: BaseRegistry, dest: WritableRegistry ) -> None: diff --git a/packages/timenet/src/timenet/registry/base.py b/packages/timenet/src/timenet/registry/base.py index 81e7be8d4..d6bd2c131 100644 --- a/packages/timenet/src/timenet/registry/base.py +++ b/packages/timenet/src/timenet/registry/base.py @@ -13,12 +13,12 @@ from typing import BinaryIO, TypeVar from timenet._fs import staged_directory -from timenet.errors import TimeFValidationError, TimeNetDatasetNotFoundError, TimeNetFileError, TimeNetRegistryError +from timenet.errors import TimeNetDatasetNotFoundError, TimeNetFileError, TimeNetRegistryError 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, validate_manifest_reference +from timenet.registry._paths import safe_version_path +from timenet.registry.dependencies import validate_manifest_reference from timenet.registry.shared import complete_copy from timenet.registry.version import DatasetVersion from timenet.sources import FoundInRegistry, NotFound, NotServed, ReleaseLookup @@ -29,7 +29,7 @@ T = TypeVar("T") -# Reports download progress: called with each file (or chunk) of bytes as it lands. A ``download_version`` +# Reports download progress: called with each file (or chunk) of bytes as it lands. A ``download_release`` # implementation calls it so a caller can render a progress bar; ``None`` downloads silently. ProgressCallback = Callable[[int], None] @@ -176,47 +176,6 @@ def resolve(self, dataset_id: str, version: Version | None) -> ReleaseLookup: return FoundInRegistry(held, self, manifest) return NotServed(held, access, access_requirements(manifest.metadata, manifest.lineage), manifest) - def download_version( # ruff: ignore[too-many-arguments] - self, - dataset_id: str, - version: str, - dest_dir: str | Path, - *, - force: bool = False, - manifest: Manifest | None = None, - progress_cb: ProgressCallback | None = None, - ) -> None: - """Download complete local copies of a release and of each of its payload sources. - - Only the release's own payload sources are downloaded: it lists every release whose files it - reads. Each is checked against the files the release pins. - - Args: - dataset_id: The dataset ID. - version: The exact release version. - dest_dir: The local ``///`` directory. - force: Download the requested release again. - manifest: An already-loaded manifest for the requested release. - progress_cb: Report transferred bytes. - - Raises: - TimeFValidationError: If the destination does not identify the requested release. - """ - reference = DatasetRef.parse(f"{dataset_id}@{version}") - root, destination_ref = split_release_dir(Path(dest_dir)) - if destination_ref != reference: - raise TimeFValidationError( - f"download destination {dest_dir} identifies {destination_ref}, expected {reference}" - ) - prepare_local_copy( - root, - reference, - fetch_from=self, - manifest=manifest, - force=force, - progress_cb=progress_cb, - ) - def download_release( self, reference: DatasetRef, diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py index 22dc22664..762d3f971 100644 --- a/packages/timenet/src/timenet/registry/local.py +++ b/packages/timenet/src/timenet/registry/local.py @@ -7,16 +7,17 @@ import pyarrow.fs as pafs -from timenet._fs import staged_directory +from timenet._fs import same_directory, staged_directory from timenet.dataset import TimeFDataset from timenet.errors import TimeFFormatError, TimeFValidationError, 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.base import BaseRegistry from timenet.registry.dependencies import check_file_sizes, validate_manifest_reference from timenet.registry.version import DatasetVersion from timenet.registry.writable import WritableRegistry -from timenet.sources import FoundInRegistry, NotFound, ReleaseLookup +from timenet.sources import FoundInRegistry, NotFound, ReleaseLookup, ReleaseSource from timenet.types import DatasetMetadata, Version from timenet.types.metadata_fields import validate_dataset_id from timenet.types.references import DatasetRef, is_latest @@ -325,6 +326,29 @@ def _latest_version(self, dataset_id: str) -> str | None: return max(versions, key=Version.parse) +def is_same_registry(source: ReleaseSource, registry: BaseRegistry) -> bool: + """Return whether a source is a given registry, or another local registry over the same directory. + + Two local registries over one directory hold the same releases. So a copy from one of them to the + other copies the directory onto itself. A build counts such a source as its output, and + :meth:`~timenet.client.TimeNet.download` counts it as the store. + + Args: + source: The source. + registry: The registry, for example the output of a build or the store of a client. + + Returns: + ``True`` for the same object or local registries over the same directory. + """ + if source is registry: + return True + return ( + isinstance(source, LocalRegistry) + and isinstance(registry, LocalRegistry) + and same_directory(source.root, registry.root) + ) + + def _is_version(name: str) -> bool: """Return whether a directory name is a parseable version. diff --git a/packages/timenet/tests/coordinator/test_build.py b/packages/timenet/tests/coordinator/test_build.py index c957ffd8c..fc866963a 100644 --- a/packages/timenet/tests/coordinator/test_build.py +++ b/packages/timenet/tests/coordinator/test_build.py @@ -8,7 +8,7 @@ from timenet.coordinator.plan import BuildStep, CopyStep, Plan from timenet.dataset import TimeFDataset from timenet.errors import TimeNetAccessError, TimeNetRegistryError, TimeNetVersionUnavailableError -from timenet.progress import StepFinished, StepStarted +from timenet.progress import BytesMoved, StepFinished, StepStarted from timenet.reader import TimeFReader from timenet.registry import LocalRegistry, open_local_version from timenet.registry._paths import release_dir @@ -144,7 +144,8 @@ def test_a_build_copies_a_missing_parent_and_reports_each_step_once(tmp_path, st CopyStep(PARENT_REF, origin), _build(CHILD_REF, depends_on=(PARENT_REF,)), ) - assert events == [ + # A copy also sends the bytes that it moves. + assert [event for event in events if not isinstance(event, BytesMoved)] == [ StepStarted(PARENT_REF, "copy"), StepFinished(PARENT_REF, "copy"), StepStarted(CHILD_REF, "build"), diff --git a/packages/timenet/tests/reader/test_shared_storage.py b/packages/timenet/tests/reader/test_shared_storage.py index 0d9a7b49b..c09ca7c1e 100644 --- a/packages/timenet/tests/reader/test_shared_storage.py +++ b/packages/timenet/tests/reader/test_shared_storage.py @@ -13,6 +13,7 @@ from timenet.manifest import Manifest from timenet.reader import LoadPlan, TimeFReader from timenet.registry import LocalRegistry, open_local_version +from timenet.registry.dependencies import prepare_local_copy from timenet.testing import make_dataset from timenet.types import ( Annotation, @@ -599,7 +600,7 @@ def test_a_download_fetches_only_the_direct_payload_sources(tmp_path): b = reader.read() store(registry_root, _derive(b, "test/c", [record for record in b.records if _name(record) == "b"])) cache = tmp_path / "cache" - LocalRegistry(registry_root).download_version("test/c", "1.0.0", cache / "test/c/1.0.0") + prepare_local_copy(cache, DatasetRef.parse("test/c@1.0.0"), fetch_from=LocalRegistry(registry_root)) assert sorted(path.parent.relative_to(cache).as_posix() for path in cache.rglob("manifest.json")) == [ "test/b/1.0.0", "test/c/1.0.0", diff --git a/packages/timenet/tests/registry/test_copy.py b/packages/timenet/tests/registry/test_copy.py index 43514741f..bb52830e4 100644 --- a/packages/timenet/tests/registry/test_copy.py +++ b/packages/timenet/tests/registry/test_copy.py @@ -17,11 +17,12 @@ from timenet.dataset import Record, Signal, Source, TimeFDataset from timenet.errors import ( TimeFFormatError, + TimeFValidationError, TimeNetAccessError, TimeNetMissingParentError, TimeNetVersionConflictError, ) -from timenet.progress import ProgressEvent, StepFinished, StepStarted +from timenet.progress import BytesMoved, 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 @@ -149,6 +150,11 @@ def _remote(tmp_path: Path, name: str, *, token: str | None = None) -> tuple[Rem return registry, root +def _steps(events: list[ProgressEvent]) -> list[ProgressEvent]: + """Return the step events, without the byte events of each copy.""" + return [event for event in events if not isinstance(event, BytesMoved)] + + @pytest.fixture def source(tmp_path) -> LocalRegistry: root = tmp_path / "source" @@ -264,7 +270,7 @@ def test_copy_release_reads_a_payload_source_that_only_dest_holds(source, dest): 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")] + assert _steps(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): @@ -321,17 +327,42 @@ def test_copy_release_repairs_a_damaged_local_copy(source, tmp_path): 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")] + assert _steps(events) == [StepStarted(child_ref, "copy"), StepFinished(child_ref, "copy")] + + +def test_copy_release_with_force_copies_a_complete_local_copy_again(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) + parent_dir = dest.root / parent_ref.dataset_id / str(parent_ref.version) + for directory in (child_dir, parent_dir): + (directory / "stale.txt").write_text("left over") + events: list[ProgressEvent] = [] + copy_release(child_ref, source, dest, force=True, progress=events.append) + # The requested release is copied again as a whole. Its complete payload source stays. + assert not (child_dir / "stale.txt").exists() + assert (parent_dir / "stale.txt").exists() + assert complete_copy(child_dir, dest.get_manifest(*child)) + assert _steps(events) == [StepStarted(child_ref, "copy"), StepFinished(child_ref, "copy")] + # force=True does not replace a copy with other files. + other = LocalRegistry(tmp_path / "other") + other.put_dataset(_other_build(child_ref.dataset_id)) + with pytest.raises(TimeNetVersionConflictError, match="new version"): + copy_release(child_ref, source, other, force=True) + + +def test_copy_release_with_force_refuses_a_registry_that_is_not_local(source, tmp_path): + dest, _ = _remote(tmp_path, "dest", token="tok_rw") + with pytest.raises(TimeFValidationError, match=r"^force=True needs a local dest"): + copy_release(child_ref, source, dest, force=True) + assert not dest.exists(*child) + assert not dest.exists(*parent) 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 == [ + assert _steps(events) == [ StepStarted(parent_ref, "copy"), StepFinished(parent_ref, "copy"), StepStarted(child_ref, "copy"), @@ -339,6 +370,29 @@ def test_copy_release_reports_each_copied_release(source, tmp_path): ] +@pytest.mark.parametrize("kind", ["local", "remote"]) +def test_copy_release_reports_the_bytes_of_each_release_that_it_copies_to_a_local_registry(kind, tmp_path): + if kind == "local": + _write_parent_and_child(tmp_path / "source") + origin: LocalRegistry | RemoteRegistry = LocalRegistry(tmp_path / "source") + else: + origin, root = _remote(tmp_path, "source") + _write_parent_and_child(root) + events: list[ProgressEvent] = [] + copy_release(child_ref, origin, LocalRegistry(tmp_path / "dest"), progress=events.append) + for release in (parent_ref, child_ref): + # The bytes of a release arrive between the start and the end of its step. + start, end = events.index(StepStarted(release, "copy")), events.index(StepFinished(release, "copy")) + moved = events[start + 1 : end] + manifest = origin.get_manifest(release.dataset_id, str(release.version)) + total = sum(part.size for part in manifest.files.all_files()) + done = [event.done for event in moved if isinstance(event, BytesMoved)] + assert done + assert moved == [BytesMoved(str(release), count, total) for count in done] + assert done == sorted(done) + assert done[-1] == total + + 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) diff --git a/packages/timenet/tests/registry/test_remote.py b/packages/timenet/tests/registry/test_remote.py index 59c182c16..bb7eb1c94 100644 --- a/packages/timenet/tests/registry/test_remote.py +++ b/packages/timenet/tests/registry/test_remote.py @@ -22,7 +22,7 @@ from timenet.format.checksums import file_checksum from timenet.manifest import Manifest from timenet.reader import LoadPlan, TimeFReader -from timenet.registry import LocalRegistry, RemoteRegistry, open_local_version +from timenet.registry import LocalRegistry, RemoteRegistry, copy_release, open_local_version from timenet.sources import FoundInRegistry, NotServed from timenet.testing import assert_datasets_equal, make_dataset from timenet.types import Access, DatasetSchema, Domain, License, Version @@ -196,13 +196,27 @@ def test_token_required_when_configured(version_dir, tmp_path): def test_full_download_writes_identical_bytes(version_dir, tmp_path): registry, _ = _remote(version_dir, tmp_path) directory, manifest = version_dir - dest = tmp_path / "out" / manifest.metadata.dataset_id / str(manifest.metadata.dataset_version) - registry.download_version(manifest.metadata.dataset_id, str(manifest.metadata.dataset_version), dest) + release = DatasetRef(dataset_id=manifest.metadata.dataset_id, version=manifest.metadata.dataset_version) + dest = tmp_path / "out" / release.dataset_id / str(release.version) + registry.download_release(release, dest) for part in manifest.files.all_parts(): assert (dest / part).read_bytes() == (directory / part).read_bytes() assert (dest / "manifest.json").exists() +@pytest.mark.parametrize("version", [None, "1.0.0"]) +def test_a_download_into_the_store_resolves_a_remote_release_once(version_dir, tmp_path, version): + registry, requests = _remote(version_dir, tmp_path) + directory, manifest = version_dir + store = LocalRegistry(tmp_path / "store") + + path = TimeNet(sources=[store, registry]).download(manifest.metadata.dataset_id, version) + + assert (path / "manifest.json").read_bytes() == (directory / "manifest.json").read_bytes() + # One read of the manifest resolves the release, and one starts the copy. + assert sum(request.url.path.endswith("/manifest") for request in requests) == 2 + + def test_full_download_waits_out_a_rate_limited_presign(version_dir, tmp_path, monkeypatch, caplog): directory, manifest = version_dir transport, _ = build_fake(directory) @@ -217,9 +231,10 @@ def handler(request): sleep = AsyncMock() monkeypatch.setattr("timenet.registry.remote._http.asyncio.sleep", sleep) registry = RemoteRegistry("http://api.local", transport=httpx.MockTransport(handler), cache_dir=tmp_path / "cache") - dest = tmp_path / "out" / manifest.metadata.dataset_id / str(manifest.metadata.dataset_version) + release = DatasetRef(dataset_id=manifest.metadata.dataset_id, version=manifest.metadata.dataset_version) + dest = tmp_path / "out" / release.dataset_id / str(release.version) - registry.download_version(manifest.metadata.dataset_id, str(manifest.metadata.dataset_version), dest) + registry.download_release(release, dest) assert limited assert sleep.await_args_list == [call(1.0)] * len(limited) @@ -239,10 +254,11 @@ def handler(request): monkeypatch.setattr("timenet.registry.remote._http.asyncio.sleep", AsyncMock()) registry = RemoteRegistry("http://api.local", transport=httpx.MockTransport(handler), cache_dir=tmp_path / "cache") - dest = tmp_path / "out" / manifest.metadata.dataset_id / str(manifest.metadata.dataset_version) + release = DatasetRef(dataset_id=manifest.metadata.dataset_id, version=manifest.metadata.dataset_version) + dest = tmp_path / "out" / release.dataset_id / str(release.version) with pytest.raises(TimeNetRegistryError, match="429"): - registry.download_version(manifest.metadata.dataset_id, str(manifest.metadata.dataset_version), dest) + registry.download_release(release, dest) assert not dest.exists() @@ -257,7 +273,7 @@ def test_a_corrupted_remote_download_names_the_file_and_its_checksums(version_di dest = tmp_path / "out" / release.dataset_id / str(release.version) with pytest.raises(TimeNetFileError) as caught: - registry.download_version(release.dataset_id, str(release.version), dest) + registry.download_release(release, dest) error = caught.value # A caller that handles format errors still catches it. @@ -279,7 +295,7 @@ def test_a_remote_download_whose_size_differs_from_its_record_names_both_sizes(v dest = tmp_path / "out" / release.dataset_id / str(release.version) with pytest.raises(TimeNetFileError) as caught: - registry.download_version(release.dataset_id, str(release.version), dest) + registry.download_release(release, dest) error = caught.value size = record["size"] @@ -485,8 +501,8 @@ def handler(request): def test_a_registry_url_in_the_sources_caches_under_the_storage_directory(tmp_path, monkeypatch): - # A remote registry that the sources name by URL caches under the storage directory, where - # download() copies to, so download() and load() share one cache-first directory. + # A remote registry that the sources name by URL reuses the complete copies under the storage + # directory. monkeypatch.setenv("TIMENET_STORAGE", str(tmp_path / "store")) (registry,) = TimeNet(sources=["https://registry.example.test"])._sources assert isinstance(registry, RemoteRegistry) @@ -545,7 +561,7 @@ def handler(request): assert any(parent_id + "/1.0.0/download/" in path for path in requests) assert not any(parent_id + "/1.0.0/download/control.duckdb" in path for path in requests) assert not (cache / parent_id / "1.0.0").exists() - remote.download_version("test/child", "1.0.0", cache / "test/child/1.0.0") + copy_release(DatasetRef.parse("test/child@1.0.0"), remote, LocalRegistry(cache)) with TimeNet(sources=[remote]).open("test/child", "1.0.0") as reader: assert reader.get(Signal, signal_id).to_arrow().equals(expected) assert not any("test/build-only" in path for path in requests) @@ -555,7 +571,9 @@ def handler(request): assert reader.get(Signal, signal_id).to_arrow().equals(expected) # The parent is already complete: preparing it again requires no network requests. prepared_requests = list(requests) - remote.download_version(parent_id, "1.0.0", cache / parent_id / "1.0.0") + parent = DatasetRef.parse(f"{parent_id}@1.0.0") + manifest = LocalRegistry(cache).get_manifest(parent_id, "1.0.0") + remote.download_release(parent, cache / parent_id / "1.0.0", manifest=manifest) assert (cache / parent_id / "1.0.0/control.duckdb").is_file() assert requests == prepared_requests @@ -578,13 +596,16 @@ def _gated_release(origin: Path, page: str) -> tuple[str, DatasetRef]: return dataset.metadata.dataset_id, release -def test_a_read_checks_access_before_it_reads_from_a_remote_registry(tmp_path): - dataset_id, release = _gated_release(tmp_path / "origin", "ancestor") +@pytest.mark.parametrize("entry_point", ["open", "download"]) +@pytest.mark.parametrize("page", ["own", "ancestor"]) +def test_a_read_or_a_download_checks_access_before_it_reads_from_a_remote_registry(tmp_path, page, entry_point): + dataset_id, release = _gated_release(tmp_path / "origin", page) transport, requests = build_fake(next((tmp_path / "origin").rglob("manifest.json")).parent) remote = RemoteRegistry("http://api.local", transport=transport, cache_dir=tmp_path / "cache") + client = TimeNet(sources=[LocalRegistry(tmp_path / "local"), remote]) with pytest.raises(TimeNetAccessError) as caught: - TimeNet(sources=[LocalRegistry(tmp_path / "local"), remote]).open(dataset_id) + getattr(client, entry_point)(dataset_id) assert caught.value.requirements == (AccessRequirement(release, Access.CREDENTIALED, "https://example.org/dua"),) assert caught.value.hint == ( diff --git a/packages/timenet/tests/registry/test_s3.py b/packages/timenet/tests/registry/test_s3.py index 4797ab5bc..a63e3a3c3 100644 --- a/packages/timenet/tests/registry/test_s3.py +++ b/packages/timenet/tests/registry/test_s3.py @@ -99,13 +99,14 @@ def test_objects_live_under_a_datasets_prefix(s3_root, tmp_path): assert all(key.startswith("registry/datasets/") for key in keys) -def test_download_then_load_serves_from_cache(s3_root, tmp_path, monkeypatch): +def test_download_then_load_serves_from_cache(s3_root, tmp_path): registry = S3Registry(s3_root, cache_dir=tmp_path / "cache") dataset_id = make_dataset().metadata.dataset_id registry.put_dataset(make_dataset()) - monkeypatch.setenv("TIMENET_STORAGE", str(tmp_path / "cache")) - client = TimeNet(sources=[registry]) - path = client.download(dataset_id) # boto3 GetObject per file, into the cache + # The store is the cache directory of the registry, so the registry reuses each download. + client = TimeNet(sources=[LocalRegistry(tmp_path / "cache"), registry]) + path = client.download(dataset_id) # boto3 GetObject per file, into the store + assert path == tmp_path / "cache" / dataset_id / "1.0.0" assert (path / "manifest.json").exists() handle = registry.open_release(DatasetRef.parse(f"{dataset_id}@1.0.0")) assert isinstance(handle.filesystem, pafs.LocalFileSystem) # cache-first: no S3 reads diff --git a/packages/timenet/tests/registry/test_version.py b/packages/timenet/tests/registry/test_version.py index d0e5c3b4b..2bfcbc81b 100644 --- a/packages/timenet/tests/registry/test_version.py +++ b/packages/timenet/tests/registry/test_version.py @@ -169,14 +169,6 @@ def test_preparation_needs_only_the_direct_payload_sources(registry_root, tmp_pa assert prepared.resolve_payload_source(SourceKey(str(parent_ref))).payload_sources == {} -@pytest.mark.parametrize("suffix", ["different/ecg/2.0.0", "demo/ecg/3.0.0", "demo/ecg/latest"]) -def test_download_rejects_a_destination_for_another_release(registry_root, tmp_path, suffix): - destination = tmp_path / "downloads" / suffix - with pytest.raises(TimeFValidationError, match=r"expected demo/ecg@2\.0\.0|release directory"): - LocalRegistry(registry_root).download_version("demo/ecg", "2.0.0", destination) - assert not destination.exists() - - @pytest.mark.parametrize("damage", [None, "missing", "truncated"]) def test_a_complete_copy_has_every_file_with_its_recorded_size(registry_root, damage): version = open_local_version(_version_dir(registry_root)) diff --git a/packages/timenet/tests/test_cli.py b/packages/timenet/tests/test_cli.py index 8e82e141a..5da0e8db3 100644 --- a/packages/timenet/tests/test_cli.py +++ b/packages/timenet/tests/test_cli.py @@ -4,6 +4,7 @@ import click import pytest +from rich.console import Console import typer from typer.testing import CliRunner @@ -13,7 +14,7 @@ from timenet.errors import TimeNetDatasetNotFoundError from timenet.reader import TimeFReader from timenet.registry import open_local_version -from timenet.testing import make_dataset +from timenet.testing import install_backends, make_dataset from timenet.types import Access, Domain from timenet.types.reference_keys import ParentAlias from timenet.types.references import DatasetRef @@ -151,12 +152,42 @@ def test_info_lists_the_access_page_of_each_release_that_sets_one(tmp_path): assert cells[start + 3][0] == "domains" -def test_download(registry_root, tmp_path, monkeypatch): +def test_download_command_copies_into_the_store_and_then_has_nothing_to_do(registry_root, tmp_path, monkeypatch): + # The last line names a long path, so the console must not wrap it. + monkeypatch.setattr(_cli_module.ui, "rich", Console(stderr=True, width=1000)) monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home")) - monkeypatch.setenv("TIMENET_STORAGE", str(tmp_path / "store")) - result = runner.invoke(app, ["download", "timenet/hello-world", "--registry", str(registry_root)]) - assert result.exit_code == 0, result.output - assert (tmp_path / "store" / "timenet/hello-world" / "1.0.0" / "manifest.json").exists() + monkeypatch.setenv("TIMENET_REGISTRY", str(tmp_path / "store")) + command = ["download", "timenet/hello-world", "--registry", str(registry_root)] + target = tmp_path / "store" / "timenet/hello-world" / "1.0.0" + + copied = runner.invoke(app, command) + again = runner.invoke(app, command) + + assert copied.exit_code == 0, copied.output + assert copied.stdout == f"{target}\n" # only the directory, for a script to capture + assert (target / "manifest.json").exists() + successes = [line for line in copied.stderr.splitlines() if line.startswith("✅")] + assert successes == ["✅ Copied 'timenet/hello-world' → 1.0.0"] + assert again.exit_code == 0, again.output + assert again.stdout == f"{target}\n" + stored = tmp_path / "store" + assert again.stderr.splitlines() == [ + f"✅ Nothing to do: 'timenet/hello-world' 1.0.0 is already in local registry {stored}" + ] + + +def test_download_command_with_force_copies_a_held_release_again(registry_root, tmp_path, monkeypatch): + install_backends(monkeypatch) + monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home")) + monkeypatch.setenv("TIMENET_REGISTRY", str(tmp_path / "store")) + command = ["download", "timenet/hello-world", "--registry", str(registry_root)] + assert runner.invoke(app, command).exit_code == 0 + + forced = runner.invoke(app, [*command, "--force"]) + + assert forced.exit_code == 0, forced.output + successes = [line for line in click.unstyle(forced.stderr).splitlines() if line.startswith("✅")] + assert successes == ["✅ Copied 'timenet/hello-world' → 1.0.0"] def test_search_rejects_unknown_filter_value(registry_root): diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py index 36ba34bd7..474c70986 100644 --- a/packages/timenet/tests/test_client.py +++ b/packages/timenet/tests/test_client.py @@ -1,3 +1,5 @@ +from pathlib import Path +import re import shutil from types import SimpleNamespace @@ -15,17 +17,20 @@ TimeFValidationError, TimeNetAccessError, TimeNetDatasetNotFoundError, + TimeNetFileError, TimeNetMissingParentError, TimeNetRegistryError, + TimeNetVersionConflictError, TimeNetVersionUnavailableError, ) from timenet.format.stored_values import StoredText from timenet.manifest.files import FilePart +from timenet.progress import BytesMoved, ProgressEvent, StepFinished, StepStarted from timenet.reader import Loading, LoadPlan, TimeFReader from timenet.registry import TIMENET_REGISTRY_URL, LocalRegistry, RemoteRegistry, open_local_version from timenet.sources import InstalledConnectors from timenet.testing import assert_datasets_equal, install_backends, make_dataset -from timenet.types import Access, DatasetMetadata, License, Version +from timenet.types import Access, Annotation, DatasetMetadata, License, Version from timenet.types.access import AccessRequirement from timenet.types.parent_dataset_card import ParentDatasetCard from timenet.types.reference_keys import ParentAlias @@ -105,7 +110,7 @@ def compose(self, raw_refs, context): return child -def test_resampled_child_reads_and_downloads_without_its_build_only_parent(tmp_path, monkeypatch): +def test_resampled_child_reads_and_downloads_without_its_build_only_parent(tmp_path): origin = LocalRegistry(tmp_path / "origin") origin.put_dataset(make_dataset()) metadata = _Resampled().metadata() @@ -124,10 +129,9 @@ def test_resampled_child_reads_and_downloads_without_its_build_only_parent(tmp_p with pytest.raises(TimeNetDatasetNotFoundError, match=reference.dataset_id): open_parents(metadata, origin) - cache = tmp_path / "download" - monkeypatch.setenv("TIMENET_STORAGE", str(cache)) - directory = TimeNet(sources=[origin]).download(metadata.dataset_id) - assert not (cache / reference.dataset_id).exists() + store = LocalRegistry(tmp_path / "store") + directory = TimeNet(sources=[store, origin]).download(metadata.dataset_id) + assert not (store.root / reference.dataset_id).exists() shutil.rmtree(origin.root) version = open_local_version(directory) assert version.manifest.parent_cards == manifest.parent_cards @@ -136,6 +140,51 @@ def test_resampled_child_reads_and_downloads_without_its_build_only_parent(tmp_p assert reader.read().records[0].signals[0].to_arrow().equals(expected) +def test_a_damaged_release_names_no_download_without_a_writable_local_registry(tmp_path): + registry, _ = _child_of_the_demo(tmp_path / "registry") + (registry.root / "test/child/1.0.0/control.duckdb").write_bytes(b"cut short") + # download() copies into a writable local registry, so a client without one has no repair to offer. + with pytest.raises(TimeNetFileError) as caught: + TimeNet(sources=[LocalRegistry(registry.root, read_only=True)]).open("test/child") + assert caught.value.hint is None + + +def test_a_damaged_release_outside_the_store_names_no_download(tmp_path): + origin, _ = _child_of_the_demo(tmp_path / "origin") + mirror = LocalRegistry(tmp_path / "mirror") + shutil.copytree(origin.root, mirror.root) + (mirror.root / "test/child/1.0.0/control.duckdb").write_bytes(b"cut short") + store = LocalRegistry(tmp_path / "store") + + # download() replaces only a copy in the store, so it cannot repair the mirror. + with pytest.raises(TimeNetFileError) as caught: + TimeNet(sources=[store, mirror, origin]).open("test/child") + assert (caught.value.release, caught.value.hint) == (DatasetRef.parse("test/child@1.0.0"), None) + + +def test_a_damaged_build_in_the_store_names_the_build_that_replaces_it(tmp_path, monkeypatch): + calls = [] + install_backends(monkeypatch, _Builder(calls)) + store = LocalRegistry(tmp_path / "store") + # A registry in the list that lacks the release, as the hosted registry lacks a local build. + empty = RemoteRegistry( + "http://api.local", + transport=httpx.MockTransport(lambda request: httpx.Response(404)), + cache_dir=tmp_path / "cache", + ) + client = TimeNet(sources=[store, empty, InstalledConnectors()]) + client.load("timenet/hello-world@1.0.0") # the build puts the release into the store + (store.root / "timenet/hello-world/1.0.0/control.duckdb").write_bytes(b"cut short") + + # No registry other than the local registry provides the release, so download() cannot replace it. + hint = f"Build it again with `timenet-build build timenet/hello-world --out {store.root} --force`." + for entry_point in (client.open, client.load, client.download): + with pytest.raises(TimeNetFileError) as caught: + entry_point("timenet/hello-world@1.0.0") + assert caught.value.hint == hint + assert len(calls) == 1 # nothing builds over the damaged copy + + @pytest.fixture(autouse=True) def _clean_env(monkeypatch): for var in ("TIMENET_HOME", "TIMENET_STORAGE", "TIMENET_CACHE", "TIMENET_REGISTRY"): @@ -151,13 +200,6 @@ def registry_root(tmp_path): return tmp_path / "reg" -@pytest.fixture -def storage(tmp_path, monkeypatch): - """The directory that ``download()`` copies into.""" - monkeypatch.setenv("TIMENET_STORAGE", str(tmp_path / "storage")) - return tmp_path / "storage" - - def _entry(source): """Describe one source of a list: a local registry by its root, another source by its text.""" if isinstance(source, LocalRegistry): @@ -201,7 +243,7 @@ def test_timenet_registry_env_selects_a_local_registry(tmp_path, monkeypatch): "installed connectors", ] assert {m.dataset_id for m in client.list()} == {"timenet/hello-world"} - assert client.download("timenet/hello-world") == tmp_path / "home" / "storage" / "timenet/hello-world" / "1.0.0" + assert client.download("timenet/hello-world") == tmp_path / "home" / "registry" / "timenet/hello-world" / "1.0.0" def test_a_client_takes_an_ordered_list_of_sources(tmp_path): @@ -290,52 +332,300 @@ def test_get_list_and_search_ask_the_first_registry_that_is_not_the_store(regist assert [metadata.dataset_id for metadata in alone.list()] == ["test/other"] -def test_download_copies_into_storage(registry_root, storage): - client = TimeNet(sources=[str(registry_root)]) - version_dir = client.download("timenet/hello-world") - assert version_dir == storage / "timenet/hello-world" / "1.0.0" - assert (version_dir / "manifest.json").exists() - assert list(version_dir.glob("time_series/part-*.parquet")) +def test_download_copies_the_release_and_its_payload_sources_into_the_store(tmp_path): + origin, _ = _child_of_the_demo(tmp_path / "origin") + store = LocalRegistry(tmp_path / "store") + # The path of origin opens a writable local registry too, but only the first one gets the copies. + path = TimeNet(sources=[store, origin.root]).download("test/child") + assert path == tmp_path / "store/test/child/1.0.0" + assert store.exists("timenet/hello-world", "1.0.0") -def test_download_reports_progress(registry_root, storage): - client = TimeNet(sources=[str(registry_root)]) - reported: list[int] = [] - client.download("timenet/hello-world", progress_cb=reported.append) - expected = sum(part.size for part in client.get("timenet/hello-world").files.all_files()) - assert reported # the callback fired - assert sum(reported) == expected # summed to the version's total byte size +def test_download_needs_a_local_registry_that_is_not_read_only(registry_root): + with pytest.raises(TimeFValidationError) as caught: + TimeNet(sources=[LocalRegistry(registry_root, read_only=True)]).download("timenet/hello-world") + assert str(caught.value) == ( + "download() needs a writable local registry to copy into. " + "Add a LocalRegistry without read_only=True to the sources" + ) -def test_manifest_file_rejects_path_traversal(): - with pytest.raises(ValidationError, match="stay within the dataset root"): - FilePart(path="../../escape.txt", checksum="sha256:" + "0" * 64, size=0) +def test_download_of_a_release_that_only_a_connector_provides_builds_nothing(tmp_path, monkeypatch): + calls = [] + install_backends(monkeypatch, _Builder(calls)) + store = LocalRegistry(tmp_path / "store") + with pytest.raises(TimeNetDatasetNotFoundError) as caught: + TimeNet(sources=[store, InstalledConnectors()]).download("timenet/hello-world") -def test_download_is_idempotent(registry_root, storage): - client = TimeNet(sources=[str(registry_root)]) + assert str(caught.value) == ( + "timenet/hello-world@1.0.0 is not built yet: only an installed connector provides it. " + "open() or load() builds it into the local registry." + ) + assert calls == [] + assert not store.root.exists() + + +def test_download_of_latest_fails_when_a_connector_declares_a_newer_version(registry_root, tmp_path, monkeypatch): + calls = [] + monkeypatch.setattr(_Builder, "VERSION", "2.0.0") + install_backends(monkeypatch, _Builder(calls)) + store = LocalRegistry(tmp_path / "store") + client = TimeNet(sources=[store, str(registry_root), InstalledConnectors()]) + + # "Latest" is the version of the connector, which no registry holds. + with pytest.raises(TimeNetDatasetNotFoundError, match=r"^timenet/hello-world@2\.0\.0 is not built yet: "): + client.download("timenet/hello-world@latest") + assert calls == [] + # Without a version, the download takes the registry, the first source that serves the dataset. + assert client.download("timenet/hello-world") == store.root / "timenet/hello-world/1.0.0" + + +def test_download_of_a_release_that_no_source_provides_names_each_source(tmp_path, monkeypatch): + monkeypatch.setattr(builders_module, "entry_points", lambda group: []) + sources = [LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "empty"), InstalledConnectors()] + + with pytest.raises(TimeNetDatasetNotFoundError) as caught: + TimeNet(sources=sources).download("x/y") + + # download() never builds, so the hint names only a registry. + names = ", ".join(str(source) for source in sources) + assert str(caught.value) == f"x/y is in none of the sources: {names}. Add a registry that holds it to the sources." + + +def test_download_of_a_version_that_the_connector_does_not_build_names_the_built_version(tmp_path, monkeypatch): + calls = [] + install_backends(monkeypatch, _Builder(calls)) + client = TimeNet(sources=[LocalRegistry(tmp_path / "store"), InstalledConnectors()]) + + with pytest.raises(TimeNetVersionUnavailableError, match=r"builds version 1\.0\.0, not the requested 9\.9\.9"): + client.download("timenet/hello-world@9.9.9") + assert calls == [] + + +def test_download_copies_each_release_from_the_first_registry_that_provides_it(tmp_path): + origin, reference = _child_of_the_demo(tmp_path / "origin") + # The first registry holds only the child, and the second only the payload source that it reads. + shutil.copytree(origin.root / "test/child", tmp_path / "first/test/child") + shutil.copytree(origin.root / reference.dataset_id, tmp_path / "second" / reference.dataset_id) + store = LocalRegistry(tmp_path / "store") + + path = TimeNet(sources=[store, tmp_path / "first", tmp_path / "second"]).download("test/child") + + assert path == store.root / "test/child/1.0.0" + with TimeNet(sources=[store]).open("test/child") as reader: + reader.verify() + + +@pytest.mark.parametrize("case", ["force", "damaged"]) +def test_download_refuses_a_store_copy_with_other_files(registry_root, tmp_path, case): + store = LocalRegistry(tmp_path / "store") + other = make_dataset() + other.records[0].annotate(Annotation(key="build", value="another")) + store.put_dataset(other) + directory = store.root / "timenet/hello-world/1.0.0" + client = TimeNet(sources=[store, str(registry_root)]) + if case == "force": + # A complete copy in the store returns at once, so only force=True compares its files. + assert client.download("timenet/hello-world") == directory + else: + (directory / "control.duckdb").write_bytes(b"cut short") + before = (directory / "manifest.json").read_text() + + with pytest.raises(TimeNetVersionConflictError, match="new version"): + client.download("timenet/hello-world", force=case == "force") + + assert (directory / "manifest.json").read_text() == before + + +@pytest.mark.parametrize("case", ["force", "damaged"]) +def test_download_names_the_store_when_no_other_registry_provides_its_copy(tmp_path, monkeypatch, case): + calls = [] + install_backends(monkeypatch, _Builder(calls)) + store = LocalRegistry(tmp_path / "store") + store.put_dataset(make_dataset()) + client = TimeNet(sources=[store, InstalledConnectors()]) + + # The store holds the release, so the error is about its copy, not about a build. + if case == "force": + message = r"^No registry other than the local registry provides timenet/hello-world@1\.0\.0, so download\(force=True\)" + with pytest.raises(TimeNetDatasetNotFoundError, match=message): + client.download("timenet/hello-world", force=True) + else: + (store.root / "timenet/hello-world/1.0.0/control.duckdb").write_bytes(b"cut short") + # The installed connector builds the release, so the hint names the build that replaces it. + hint = f"Build it again with `timenet-build build timenet/hello-world --out {store.root} --force`." + with pytest.raises(TimeNetFileError) as caught: + client.download("timenet/hello-world") + assert caught.value.hint == hint + assert calls == [] + + +@pytest.mark.parametrize("case", ["force", "damaged"]) +def test_download_counts_a_registry_over_the_directory_of_the_store_as_the_store(tmp_path, monkeypatch, case): + monkeypatch.setattr(builders_module, "entry_points", lambda group: []) + root = tmp_path / "reg" + LocalRegistry(root).put_dataset(make_dataset()) + directory = root / "timenet/hello-world/1.0.0" + if case == "damaged": + (directory / "control.duckdb").write_bytes(b"cut short") + before = {path: path.read_bytes() for path in directory.rglob("*") if path.is_file()} + # As for `TIMENET_REGISTRY=./reg timenet download timenet/hello-world --registry ./reg`: the + # registry after the store reads the directory of the store, so it cannot replace its copy. + monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home")) + monkeypatch.setenv("TIMENET_REGISTRY", str(root)) + client = TimeNet(sources=default_sources(remote_registry=root)) + + if case == "force": + with pytest.raises(TimeNetDatasetNotFoundError) as caught: + client.download("timenet/hello-world", force=True) + assert str(caught.value) == ( + "No registry other than the local registry provides timenet/hello-world@1.0.0, so download(force=True) " + "cannot copy it again." + ) + else: + with pytest.raises(TimeNetFileError) as caught: + client.download("timenet/hello-world") + assert ( + caught.value.hint + == "No registry other than the local registry provides it, so download() cannot replace it." + ) + assert {path: path.read_bytes() for path in directory.rglob("*") if path.is_file()} == before + + +def test_download_counts_a_case_variant_of_the_store_directory_as_the_store(tmp_path): + (tmp_path / "Probe").mkdir() + if not (tmp_path / "probe").exists(): + pytest.skip("the temporary folder tells upper and lower case apart") + store = LocalRegistry(tmp_path / "store") + store.put_dataset(make_dataset()) + client = TimeNet(sources=[store, LocalRegistry(tmp_path / "STORE")]) + + # The second registry reads the directory of the store, so it cannot copy the release again. + with pytest.raises(TimeNetDatasetNotFoundError, match=r"^No registry other than the local registry provides "): + client.download("timenet/hello-world", force=True) + + +@pytest.mark.parametrize("case", ["missing", "damaged"]) +def test_download_completes_the_payload_sources_of_a_complete_store_copy(tmp_path, case): + origin, reference = _child_of_the_demo(tmp_path / "origin") + store = LocalRegistry(tmp_path / "store") + client = TimeNet(sources=[store, origin]) + client.download("test/child") + source = store.root / reference.dataset_id / str(reference.version) + if case == "missing": + shutil.rmtree(source) + else: + (source / "control.duckdb").write_bytes(b"cut short") + + # The child is complete in the store, but the payload source that it reads is not. + assert client.download("test/child") == store.root / "test/child/1.0.0" + with TimeFReader(store.open_release(reference)) as reader: + reader.verify() + + +def _store_reading(registry: LocalRegistry, release: DatasetRef, source: DatasetRef, record: int) -> None: + """Store ``release``, which keeps one Record of ``source`` and adds one Record of its own values.""" + dataset = TimeFDataset( + metadata=make_dataset().metadata.model_copy( + update={"dataset_id": release.dataset_id, "parents": {ParentAlias("source"): source}} + ) + ) + with TimeFReader(open_local_version(registry.root / source.dataset_id / str(source.version))) as reader: + dataset.import_from(reader, alias=ParentAlias("source"), records=[record]) + own = make_dataset().records[0].signals[0] + signal = Signal.from_values(own.to_numpy(), spec=own.spec, name="own", time_axis=RegularAxis.from_rate_hz(16)) + dataset.add_record(record=Record(sources=(Source(name="Monitor", signals=(signal,)),))) + registry.put_dataset(dataset) + + +def test_download_copies_a_payload_source_without_the_payload_sources_of_that_source(tmp_path): + # The store holds READER, which reads the own Record of SOURCE. SOURCE reads DEMO. Origin holds + # SOURCE without DEMO, which READER does not need. + work = LocalRegistry(tmp_path / "work") + work.put_dataset(make_dataset()) + demo = DatasetRef(dataset_id=make_dataset().metadata.dataset_id, version=make_dataset().metadata.dataset_version) + source, reader = DatasetRef.parse("test/source@1.0.0"), DatasetRef.parse("test/reader@1.0.0") + _store_reading(work, source, demo, make_dataset().records[0].id) + _store_reading(work, reader, source, 2) + assert [pinned.release for pinned in work.get_manifest(reader.dataset_id).payload_sources.values()] == [source] + store = LocalRegistry(tmp_path / "store") + origin = LocalRegistry(tmp_path / "origin") + for registry, release in ((store, reader), (origin, source)): + path = Path(release.dataset_id, str(release.version)) + shutil.copytree(work.root / path, registry.root / path) + + assert TimeNet(sources=[store, origin]).download(reader.dataset_id) == store.root / "test/reader/1.0.0" + assert store.exists(source.dataset_id, str(source.version)) + assert not store.exists(demo.dataset_id, str(demo.version)) + + +def test_download_repairs_a_damaged_store_copy(tmp_path): + origin, reference = _child_of_the_demo(tmp_path / "origin") + store = LocalRegistry(tmp_path / "store") + client = TimeNet(sources=[store, origin.root]) + client.download("test/child") + (store.root / reference.dataset_id / str(reference.version) / "control.duckdb").write_bytes(b"cut short") + + # The read stops at the damaged payload source, and names the download that repairs it. + hint = f"Copy an intact release into the local registry with download('{reference}')." + with pytest.raises(TimeNetFileError, match=re.escape(hint) + "$"): + client.open("test/child") + client.download(str(reference)) + + with client.open("test/child") as reader: + reader.verify() + + +def test_download_returns_a_complete_store_copy_at_once(registry_root, tmp_path): + client = TimeNet(sources=[LocalRegistry(tmp_path / "store"), str(registry_root)]) first = client.download("timenet/hello-world") - second = client.download("timenet/hello-world") - assert first == second + events: list[ProgressEvent] = [] + assert client.download("timenet/hello-world", progress=events.append) == first + assert events == [] # nothing is copied again -def test_force_redownload_replaces_atomically(registry_root, storage): - client = TimeNet(sources=[str(registry_root)]) +def test_download_with_force_copies_again(registry_root, tmp_path): + client = TimeNet(sources=[LocalRegistry(tmp_path / "store"), str(registry_root)]) first = client.download("timenet/hello-world") - (first / "stale.txt").write_text("left over") # a marker that a clean replace should remove + (first / "stale.txt").write_text("left over") # a marker that a clean replace removes again = client.download("timenet/hello-world", force=True) assert again == first assert (again / "manifest.json").exists() - assert not (again / "stale.txt").exists() # replaced wholesale, not written in place - assert not list(again.parent.glob("*.tmp-*")) # staging dir cleaned up + assert not (again / "stale.txt").exists() # replaced as a whole, not written in place + assert not list(again.parent.glob("*.tmp-*")) # the staging directory is gone -def test_download_leaves_sibling_staging_dirs_untouched(registry_root, storage): - client = TimeNet(sources=[str(registry_root)]) +def test_download_reports_its_steps_and_bytes(tmp_path): + origin, reference = _child_of_the_demo(tmp_path / "origin") + child = DatasetRef.parse("test/child@1.0.0") + events: list[ProgressEvent] = [] + TimeNet(sources=[LocalRegistry(tmp_path / "store"), origin.root]).download("test/child", progress=events.append) + + assert [event for event in events if not isinstance(event, BytesMoved)] == [ + StepStarted(reference, "copy"), + StepFinished(reference, "copy"), + StepStarted(child, "copy"), + StepFinished(child, "copy"), + ] + for release in (reference, child): + done = [event.done for event in events if isinstance(event, BytesMoved) and event.subject == str(release)] + manifest = origin.get_manifest(release.dataset_id, str(release.version)) + assert done[-1] == sum(part.size for part in manifest.files.all_files()) + + +def test_manifest_file_rejects_path_traversal(): + with pytest.raises(ValidationError, match="stay within the dataset root"): + FilePart(path="../../escape.txt", checksum="sha256:" + "0" * 64, size=0) + + +def test_download_leaves_sibling_staging_dirs_untouched(registry_root, tmp_path): + store = LocalRegistry(tmp_path / "store") + client = TimeNet(sources=[store, str(registry_root)]) # A concurrent download of the same version has a live .tmp-* dir. download() must not # delete a sibling staging dir: sweeping siblings would corrupt that other download mid-write. - sibling = storage / "timenet/hello-world" / "1.0.0.tmp-deadbeef" + sibling = store.root / "timenet/hello-world" / "1.0.0.tmp-deadbeef" sibling.mkdir(parents=True) (sibling / "inflight.parquet").write_text("partial") @@ -391,9 +681,9 @@ def counting_close(reader): assert_datasets_equal(expected, restored) -def test_load_reads_in_place_without_downloading(registry_root, tmp_path, storage, monkeypatch): +def test_load_reads_in_place_without_downloading(registry_root, tmp_path, monkeypatch): # load() reads the release where it is, not download-then-read, so it must not write a copy into - # the store or into local storage the way download() does. + # the store the way download() does. calls = [] install_backends(monkeypatch, _Builder(calls)) store = tmp_path / "store" @@ -401,7 +691,6 @@ def test_load_reads_in_place_without_downloading(registry_root, tmp_path, storag client.load("timenet/hello-world") assert calls == [] assert not store.exists() # nothing copied into the store - assert not (storage / "timenet/hello-world").exists() # nothing fetched into the download cache def test_open_torch(registry_root, tmp_path, monkeypatch): @@ -489,9 +778,9 @@ def recording_open(client, *args, **kwargs): assert str(reader.metadata.dataset_version) == "1.0.0" -def test_version_ref_download_pins(versioned_registry, storage): - client = TimeNet(sources=[str(versioned_registry)]) - assert client.download("timenet/hello-world@1.0.0") == storage / "timenet/hello-world" / "1.0.0" +def test_version_ref_download_pins(versioned_registry, tmp_path): + client = TimeNet(sources=[LocalRegistry(tmp_path / "store"), str(versioned_registry)]) + assert client.download("timenet/hello-world@1.0.0") == tmp_path / "store/timenet/hello-world/1.0.0" def test_version_ref_missing_pin_raises(versioned_registry): @@ -590,9 +879,9 @@ def get_manifest(self, dataset_id, version=None): def test_load_rejects_a_credentialed_dataset_from_a_hosted_registry(tmp_path): - client = TimeNet( - sources=[_hosted(tmp_path, _card("org/gated", Access.CREDENTIALED, "https://physionet.example/dua"))] - ) + hosted = _hosted(tmp_path, _card("org/gated", Access.CREDENTIALED, "https://physionet.example/dua")) + store = LocalRegistry(tmp_path / "store") + client = TimeNet(sources=[store, hosted]) message = ( "org/gated@1.0.0 is credentialed. Get access at https://physionet.example/dua for org/gated@1.0.0. " "No source serves its files. Add InstalledConnectors() to the sources, or build it locally with " @@ -601,6 +890,7 @@ def test_load_rejects_a_credentialed_dataset_from_a_hosted_registry(tmp_path): with pytest.raises(TimeNetAccessError) as caught: client.load("org/gated") assert str(caught.value) == message + assert not store.root.exists() def test_the_access_hint_says_that_no_installed_connector_builds_the_release(tmp_path, monkeypatch): diff --git a/packages/timenet/tests/test_fs.py b/packages/timenet/tests/test_fs.py index 6fcd240b8..c381cdb9b 100644 --- a/packages/timenet/tests/test_fs.py +++ b/packages/timenet/tests/test_fs.py @@ -2,7 +2,24 @@ import pytest -from timenet._fs import publish_directory +from timenet._fs import publish_directory, same_directory + + +def test_a_symlink_and_its_target_are_the_same_directory(tmp_path): + (tmp_path / "store").mkdir() + (tmp_path / "link").symlink_to(tmp_path / "store", target_is_directory=True) + assert same_directory(tmp_path / "link", tmp_path / "store") + + +def test_two_missing_paths_that_differ_only_in_case_are_different_directories(tmp_path): + assert not same_directory(tmp_path / "store", tmp_path / "Store") + + +def test_a_missing_path_is_never_the_same_directory_as_an_existing_one(tmp_path): + (tmp_path / "store").mkdir() + if (tmp_path / "Store").exists(): + pytest.skip("the temporary folder ignores case, so the case variant exists too") + assert not same_directory(tmp_path / "Store", tmp_path / "store") def _directories(tmp_path: Path) -> tuple[Path, Path]: From 201f9b1b19ea6611bed1c9ed28a373f2c3ddd521 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Wed, 4 Nov 2026 19:12:59 +0100 Subject: [PATCH 2/5] docs: shorten the docstrings and docs of the download into the store Say only what a caller of download() and copy_release(force=...) needs to know. Part of TNET-157. --- packages/timenet/src/timenet/client.py | 30 +++++-------------- .../timenet/src/timenet/registry/_copy.py | 14 +++------ .../timenet/src/timenet/registry/local.py | 7 ++--- 3 files changed, 15 insertions(+), 36 deletions(-) diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index 265a00155..c45859fae 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -1,13 +1,8 @@ """The :class:`TimeNet` class is the SDK's single entry point for using TimeNet from code. -A client reads releases through an ordered list of sources: registries and -``InstalledConnectors()`` (see :mod:`timenet.sources`). Its methods ``open``, ``load``, and -``open_torch`` read a release where it is, through the read coordinator in -:mod:`timenet.coordinator.read`. If only ``InstalledConnectors()`` provides a release, they build it -into the first :class:`~timenet.registry.LocalRegistry` of the list that is not -``read_only=True``. The method ``download`` copies a release into that local registry, from the first -registry of the list that provides it. The methods ``list``, ``get``, and ``search`` ask one registry of the -list. :func:`default_sources` makes the list of ``TimeNet()``. +A client reads releases through an ordered list of sources, and builds a release that only +``InstalledConnectors()`` provides into its first local registry that is not read-only. ``download`` +copies a release into that local registry, and :func:`default_sources` makes the list of ``TimeNet()``. """ # The public API has a method named ``list``. Deferred annotations keep the type hint @@ -249,7 +244,7 @@ def download( force: bool = False, progress: ProgressCallback | None = None, ) -> Path: - """Copy a release and its payload sources into the store, from the first source that provides it. + """Copy a release and its payload sources into the local registry. The release resolves as in :meth:`open`, without a build. The release and each payload source that the local registry lacks or holds damaged come from the first registry other than the local registry that @@ -259,8 +254,7 @@ def download( dataset_id: The dataset ID. version: An exact version, ``"latest"``, or ``None``, as in :meth:`open`. force: Copy the release again, even if the local registry holds a complete copy. - progress: Receives the progress events of the copy: a step for each release that it copies - or repairs, and the bytes of each such release as they arrive. + progress: Receives a step event for each copied release, and its bytes as they arrive. Returns: The release directory in the local registry. @@ -431,9 +425,6 @@ def _catalog(self) -> BaseRegistry: def _file_hint(self, error: TimeNetFileError) -> str | None: """Tell the caller how to replace the damaged copy that an error names, if the store holds it. - :meth:`download` replaces only a copy in the store. So a damaged copy in another local - registry gets no hint. - Args: error: The error that names the damaged release. @@ -514,9 +505,7 @@ def _access_hint(dataset_id: str, sources: Sequence[ReleaseSource]) -> str: def _copy_source(sources: Sequence[ReleaseSource], store: LocalRegistry, release: DatasetRef) -> Found | None: - """Return the answer for a release of the first source other than the store that provides it. - - :meth:`TimeNet.download` copies the release only if a registry other than the store gave this answer. + """Return the answer for a release, among the sources other than the store. Args: sources: The sources of the client. @@ -524,8 +513,7 @@ def _copy_source(sources: Sequence[ReleaseSource], store: LocalRegistry, release release: The exact release. Returns: - The answer of the source that serves or builds the release, or ``None`` if no source other - than the store provides it. + The answer of the source that serves or builds it, or ``None``. """ others = [source for source in sources if not is_same_registry(source, store)] answer = resolve_release(others, release.dataset_id, str(release.version)) @@ -541,9 +529,7 @@ def _repair_hint(release: DatasetRef, found: Found | None, store: LocalRegistry) store: The local registry. Returns: - The hint as a full sentence. It names :meth:`TimeNet.download` if a registry other than the - store provides the release, else the build that replaces the copy, if an installed connector - builds the release. + The hint as a full sentence: a download from a registry, else a new build. """ if isinstance(found, FoundInRegistry) and not is_same_registry(found.registry, store): # A registry other than the local registry provides the release. diff --git a/packages/timenet/src/timenet/registry/_copy.py b/packages/timenet/src/timenet/registry/_copy.py index 6b2c71f9c..415947810 100644 --- a/packages/timenet/src/timenet/registry/_copy.py +++ b/packages/timenet/src/timenet/registry/_copy.py @@ -40,19 +40,13 @@ def copy_release( # ruff: ignore[too-many-arguments] dest: The registry to copy to. with_payload_sources: Also copy the payload sources that the manifest of the release lists. force: Fetch ``reference`` again, even if a local ``dest`` holds a complete copy of it. - progress: Receives :class:`~timenet.progress.StepStarted` and - :class:`~timenet.progress.StepFinished` for each release that is copied or repaired. For - each release that a local ``dest`` gets, it also receives - :class:`~timenet.progress.BytesMoved` events between the two, as the files arrive. Their - ``url`` is the release, and their ``total`` is the sum of the file sizes in its manifest. + progress: Receives the step events of each copied release, and its bytes for a local ``dest``. Raises: TimeNetRegistryError: If ``dest`` is read-only. TimeFValidationError: If ``force`` is set and ``dest`` is not local. - TimeNetVersionConflictError: If ``dest`` holds the release, or a payload source of it, with - other files or another card. - TimeNetAccessError: If a release of the copy is not open, and ``dest`` requires open access - or ``source`` requires it for a release that the copy fetches. + TimeNetVersionConflictError: If ``dest`` holds a release of the copy with other files or another card. + TimeNetAccessError: If a release is not open, and ``dest`` or ``source`` requires open access. TimeNetDatasetNotFoundError: If ``source`` does not hold the release. TimeNetMissingParentError: If ``source`` lacks a payload source and ``dest`` holds no complete copy of it. TimeFFormatError: If a manifest names a different release or a pin fails. @@ -123,7 +117,7 @@ def _copy_one( # ruff: ignore[too-many-arguments] def _bytes_of(reference: DatasetRef, manifest: Manifest, progress: ProgressCallback) -> Callable[[int], None]: - """Turn the byte counts of the download of one release into :class:`~timenet.progress.BytesMoved` events. + """Turn the byte counts of a release download into :class:`~timenet.progress.BytesMoved` events. Args: reference: The release. diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py index 762d3f971..e2dab398f 100644 --- a/packages/timenet/src/timenet/registry/local.py +++ b/packages/timenet/src/timenet/registry/local.py @@ -327,11 +327,10 @@ def _latest_version(self, dataset_id: str) -> str | None: def is_same_registry(source: ReleaseSource, registry: BaseRegistry) -> bool: - """Return whether a source is a given registry, or another local registry over the same directory. + """Return whether a source is a registry, or a local registry over the same directory. - Two local registries over one directory hold the same releases. So a copy from one of them to the - other copies the directory onto itself. A build counts such a source as its output, and - :meth:`~timenet.client.TimeNet.download` counts it as the store. + Two local registries over one directory hold the same releases, so a copy between them would + copy the directory onto itself. Args: source: The source. From 1219e12d783891f5c38bc0a2c5be0857cc1fa448 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Thu, 5 Nov 2026 00:27:17 +0100 Subject: [PATCH 3/5] docs: shorten the docstrings of the download into the store Shorter docstrings for the code that this change adds or changes. Part of TNET-157. --- packages/timenet/src/timenet/_fs.py | 9 ++-- packages/timenet/src/timenet/cache.py | 15 +++---- packages/timenet/src/timenet/cli/app.py | 5 ++- packages/timenet/src/timenet/client.py | 43 +++++++++---------- .../timenet/src/timenet/coordinator/plan.py | 21 +++++---- .../timenet/src/timenet/coordinator/read.py | 2 +- packages/timenet/src/timenet/progress.py | 2 +- .../timenet/src/timenet/registry/_copy.py | 38 ++++++++-------- .../timenet/src/timenet/registry/local.py | 9 ++-- 9 files changed, 69 insertions(+), 75 deletions(-) diff --git a/packages/timenet/src/timenet/_fs.py b/packages/timenet/src/timenet/_fs.py index 4580bd87a..27856a475 100644 --- a/packages/timenet/src/timenet/_fs.py +++ b/packages/timenet/src/timenet/_fs.py @@ -1,4 +1,4 @@ -"""Compare directories, and replace a directory with a complete staging directory.""" +"""Compare directory paths and replace directories from staging.""" from collections.abc import Generator from contextlib import contextmanager @@ -10,16 +10,15 @@ def same_directory(a: Path, b: Path) -> bool: """Check whether two paths name the same directory, as the file system sees them. - Two existing directories compare by their identity on disk, so a symlink or a case variant on a volume that - ignores case matches. An existing path never matches a missing one. Two missing paths match if their nearest - existing parents are the same directory and the names below those parents are equal as written. + Existing paths match when they identify the same directory on disk, and never match a missing path. + Missing paths match when their nearest existing parents match and the names below them are equal as written. Args: a: The first path, whether or not it exists. b: The second path, whether or not it exists. Returns: - ``True`` if both paths name the same directory. + ``True`` if the paths match. """ if a.exists() or b.exists(): # On a volume that ignores case, a case variant of an existing path exists too. diff --git a/packages/timenet/src/timenet/cache.py b/packages/timenet/src/timenet/cache.py index 83f6c9406..2649ab31d 100644 --- a/packages/timenet/src/timenet/cache.py +++ b/packages/timenet/src/timenet/cache.py @@ -47,10 +47,10 @@ def human_bytes(n: int) -> str: def cached_datasets() -> list[CachedDataset]: - """List every dataset version in the local registry and download storage, with sizes. + """List dataset versions and their sizes in the local registry and storage directory. Returns: - The cached datasets, sorted by (location, id, version). + The cached versions sorted by location, dataset ID, and version. """ config = settings() # Glue: the storage directory, as the note at the top of this module says. @@ -69,18 +69,15 @@ def raw_cache_size() -> int: def clear_cache(*, include_registry: bool) -> tuple[int, list[Path]]: - """Delete cached data and return the bytes freed and the directories removed. + """Delete the storage directory and raw download cache. - This always clears the download storage and the raw download cache. The caller can fetch both again. - It removes the local registry (locally built datasets) only when the caller sets ``include_registry``. + Read or removal failures raise ``OSError``. Args: - include_registry: Also remove the local registry. + include_registry: Remove the local registry too. Returns: - A ``(bytes_freed, removed_dirs)`` pair. Every directory in ``removed_dirs`` is gone when this - returns. This raises an ``OSError`` if it cannot fully remove a tree. This stops the caller from - reporting space it never freed. + A ``(bytes_freed, removed_dirs)`` pair with the removed directory paths. """ config = settings() # Glue: the storage directory, as the note at the top of this module says. diff --git a/packages/timenet/src/timenet/cli/app.py b/packages/timenet/src/timenet/cli/app.py index 209c42639..7e97f68ad 100644 --- a/packages/timenet/src/timenet/cli/app.py +++ b/packages/timenet/src/timenet/cli/app.py @@ -272,7 +272,10 @@ def cache_clear( all_: bool = typer.Option(False, "--all", help="Also remove the default local registry (builds and downloads)."), yes: bool = typer.Option(False, "--yes", "-y", help="Skip the confirmation prompt."), ) -> None: - """Delete cached data: the storage directory and the raw cache. ``--all`` also removes the default local registry.""" + """Delete the storage directory and raw cache. + + Use ``--all`` to also remove the default local registry. + """ if not yes: what = "the storage directory and the raw cache" if all_: diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index c45859fae..c67edadd1 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -1,8 +1,7 @@ -"""The :class:`TimeNet` class is the SDK's single entry point for using TimeNet from code. +"""Find, read, and download datasets with :class:`TimeNet`. -A client reads releases through an ordered list of sources, and builds a release that only -``InstalledConnectors()`` provides into its first local registry that is not read-only. ``download`` -copies a release into that local registry, and :func:`default_sources` makes the list of ``TimeNet()``. +Clients search their sources in order. Builds and downloads go into the first local registry that is +not read-only. """ # The public API has a method named ``list``. Deferred annotations keep the type hint @@ -244,17 +243,17 @@ def download( force: bool = False, progress: ProgressCallback | None = None, ) -> Path: - """Copy a release and its payload sources into the local registry. + """Copy a dataset release and its payload sources into the local registry. The release resolves as in :meth:`open`, without a build. The release and each payload source that the local registry lacks or holds damaged come from the first registry other than the local registry that provides them. Args: - dataset_id: The dataset ID. + dataset_id: The dataset ID, optionally followed by ``@version``. version: An exact version, ``"latest"``, or ``None``, as in :meth:`open`. force: Copy the release again, even if the local registry holds a complete copy. - progress: Receives a step event for each copied release, and its bytes as they arrive. + progress: Receives step events and byte counts for each copied release. Returns: The release directory in the local registry. @@ -264,10 +263,10 @@ def download( TimeNetAccessError: If a required release is only listed or a remote release is not open. TimeNetDatasetNotFoundError: If no registry other than the local registry provides the release. TimeNetVersionUnavailableError: If the version is missing and its connector builds another version. - TimeNetMissingParentError: If no registry provides a payload source that the store lacks. + TimeNetMissingParentError: If no registry provides a missing payload source. TimeNetVersionConflictError: If the local registry holds a release of the copy with different files. TimeNetFileError: If a copied file is damaged or no registry can replace a damaged local copy. - TimeFFormatError: If a payload source differs from its recorded files or the payload sources form a cycle. + TimeFFormatError: If payload sources have conflicting file records or form a dependency cycle. """ # ruff: ignore[docstring-extraneous-exception] (raised by the plan and the copies) dataset_id, version = _resolve_ref(dataset_id, version) store = self._store @@ -405,13 +404,13 @@ def load(self, dataset_id: str, version: str | None = None) -> TimeFDataset: return reader.read(load=LoadPlan.ALL_EAGER) def _catalog(self) -> BaseRegistry: - """Return the registry that :meth:`list`, :meth:`get`, and :meth:`search` ask. + """Select a registry from the sources. Returns: - The first registry other than the local registry, or the local registry if it is the only one. + The first registry other than the local registry, or the local registry if no other registry exists. Raises: - TimeFValidationError: If the sources contain no registry. + TimeFValidationError: If no source is a registry. """ # TODO(adrian): Remove temporary glue which maps these methods onto one registry until # list() and search() ask every source. This registry gives the results of the client of @@ -423,7 +422,7 @@ def _catalog(self) -> BaseRegistry: return registry def _file_hint(self, error: TimeNetFileError) -> str | None: - """Tell the caller how to replace the damaged copy that an error names, if the store holds it. + """Suggest how to replace a damaged release in the local registry. Args: error: The error that names the damaged release. @@ -505,15 +504,15 @@ def _access_hint(dataset_id: str, sources: Sequence[ReleaseSource]) -> str: def _copy_source(sources: Sequence[ReleaseSource], store: LocalRegistry, release: DatasetRef) -> Found | None: - """Return the answer for a release, among the sources other than the store. + """Find the answer of a release source outside the local registry. Args: - sources: The sources of the client. - store: The local registry. - release: The exact release. + sources: The sources to search, in order. + store: The local registry to exclude. + release: The dataset ID and exact version. Returns: - The answer of the source that serves or builds it, or ``None``. + The answer of the source that provides or builds the release, or ``None`` if none exists. """ others = [source for source in sources if not is_same_registry(source, store)] answer = resolve_release(others, release.dataset_id, str(release.version)) @@ -521,15 +520,15 @@ def _copy_source(sources: Sequence[ReleaseSource], store: LocalRegistry, release def _repair_hint(release: DatasetRef, found: Found | None, store: LocalRegistry) -> str: - """Tell the caller how to replace a damaged copy of a release in the store. + """Suggest how to replace a damaged release in the local registry. Args: - release: The release. - found: The answer that :func:`_copy_source` returns for the release. + release: The damaged release. + found: The answer of the available source, or ``None`` if none exists. store: The local registry. Returns: - The hint as a full sentence: a download from a registry, else a new build. + A message with a download or build command, or the reason replacement is unavailable. """ if isinstance(found, FoundInRegistry) and not is_same_registry(found.registry, store): # A registry other than the local registry provides the release. diff --git a/packages/timenet/src/timenet/coordinator/plan.py b/packages/timenet/src/timenet/coordinator/plan.py index bc407256d..4140ec3cc 100644 --- a/packages/timenet/src/timenet/coordinator/plan.py +++ b/packages/timenet/src/timenet/coordinator/plan.py @@ -207,7 +207,7 @@ class _Graph: open_only: bool """True if each release that the walk visits must declare ``access: open``.""" builds: bool = True - """True if the plan may build a release that only installed connectors provide.""" + """Whether the plan allows builds from installed connectors.""" releases: dict[DatasetRef, _Release] = field(default_factory=dict) """The nodes, in the order in which the walk found them.""" pins: dict[DatasetRef, dict[DatasetRef, PayloadSource]] = field(default_factory=dict) @@ -370,10 +370,10 @@ def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry, *, for reference: The exact release to copy. role: Whether the release is a dataset or a payload source. registry: The registry to copy from. - force: Copy the release again, also if the output holds a complete copy. + force: Copy the release again, even if the output holds a complete copy. Raises: - TimeNetDatasetNotFoundError: If the release is absent from the registry. + TimeNetDatasetNotFoundError: If the registry lacks the release. TimeFFormatError: If a payload source that the copy brings or reuses fails a file pin. """ # ruff: ignore[docstring-extraneous-exception] - raised by find_payload_sources manifest = self.lookups.manifest(registry, reference) @@ -593,19 +593,19 @@ def plan_steps( force_build: bool, open_access_only: bool, ) -> tuple[DatasetRef, list[Step]]: - """Plan a dataset build and all required releases. + """Plan the steps to build a dataset and its dependencies. A damaged copy of the requested release in the output gets a forced build. Args: builder: The installed connector that builds the dataset. - output: The registry to write built releases to. + output: The destination registry. parents_from: The sources to search for missing releases, in order. - force_build: Rebuild the requested release even if the output holds it. + force_build: Rebuild the requested release, even if the output holds it. open_access_only: Require open access for every release. Returns: - The requested release and ordered steps, ending with its build if needed. + The requested release and its steps, with its own build last if needed. """ card = builder.card if open_access_only: @@ -650,8 +650,8 @@ def plan_download( Args: release: The exact release. - store: The store to copy into. - sources: The sources, in order. The store, and each registry over its directory, do not count. + store: The local registry to copy into. + sources: The sources to search in order, excluding registries at the directory of ``store``. force: Copy the release again, even if the local registry holds a complete copy. lookups: The answers and manifests that resolved ``release``, or ``None`` to ask the sources. @@ -659,8 +659,7 @@ def plan_download( The copy steps, in the order in which they run. Raises: - TimeNetDatasetNotFoundError: If only installed connectors provide a release that the store - lacks, or if ``force`` is set and only the local registry holds the release. + TimeNetDatasetNotFoundError: If a needed release requires a build or no registry can supply a forced copy. TimeNetFileError: If the local registry holds a damaged copy that no other registry provides. TimeNetAccessError: If a source lists a release that the local registry lacks but no source serves it. TimeNetMissingParentError: If no registry provides a payload source that the local registry lacks. diff --git a/packages/timenet/src/timenet/coordinator/read.py b/packages/timenet/src/timenet/coordinator/read.py index 287ff7f5d..560dcd0fd 100644 --- a/packages/timenet/src/timenet/coordinator/read.py +++ b/packages/timenet/src/timenet/coordinator/read.py @@ -97,7 +97,7 @@ def raise_not_found(dataset_id: str, version: str | None, sources: Sequence[Rele sources: The sources searched for the release. Raises: - TimeNetVersionUnavailableError: If the installed connector declares another version than the exact request. + TimeNetVersionUnavailableError: If the installed connector declares a different version. TimeNetInvalidCardError: If the card of the connector cannot be read. TimeNetDatasetNotFoundError: Otherwise, with a message that names the searched sources. """ # ruff: ignore[docstring-extraneous-exception] - raised by declared diff --git a/packages/timenet/src/timenet/progress.py b/packages/timenet/src/timenet/progress.py index 5ce2a2593..d05ce2713 100644 --- a/packages/timenet/src/timenet/progress.py +++ b/packages/timenet/src/timenet/progress.py @@ -39,7 +39,7 @@ class StepFinished: @dataclass(frozen=True) class BytesMoved: - """The byte count of a transfer in progress. + """The byte count for a transfer in progress. Attributes: subject: What moves: a source or target URL, or a copied release ID such as ``"org/name@1.0.0"``. diff --git a/packages/timenet/src/timenet/registry/_copy.py b/packages/timenet/src/timenet/registry/_copy.py index 415947810..15fc05b99 100644 --- a/packages/timenet/src/timenet/registry/_copy.py +++ b/packages/timenet/src/timenet/registry/_copy.py @@ -29,27 +29,27 @@ def copy_release( # ruff: ignore[too-many-arguments] force: bool = False, progress: ProgressCallback | None = None, ) -> None: - """Copy a release and its required payload sources to another registry. + """Copy a release and its required payload sources to a registry. - The copy reuses complete releases in ``dest`` and fetches damaged local copies again. - It checks access, file records and pins before copying payload sources and then the requested release. + Payload sources go first, a complete copy in ``dest`` is reused, and a damaged local copy is fetched again. + All checks run before the first copy, so a failed check copies nothing. 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. - force: Fetch ``reference`` again, even if a local ``dest`` holds a complete copy of it. - progress: Receives the step events of each copied release, and its bytes for a local ``dest``. + force: Fetch ``reference`` again, even if ``dest`` has a complete local copy. + progress: Receives step events for each copied release and byte counts when ``dest`` is local. Raises: TimeNetRegistryError: If ``dest`` is read-only. TimeFValidationError: If ``force`` is set and ``dest`` is not local. TimeNetVersionConflictError: If ``dest`` holds a release of the copy with other files or another card. - TimeNetAccessError: If a release is not open, and ``dest`` or ``source`` requires open access. - TimeNetDatasetNotFoundError: If ``source`` does not hold the release. - TimeNetMissingParentError: If ``source`` lacks a payload source and ``dest`` holds no complete copy of it. - TimeFFormatError: If a manifest names a different release or a pin fails. + TimeNetAccessError: If a release is not open and ``dest`` or ``source`` requires open access. + TimeNetDatasetNotFoundError: If ``source`` lacks the release. + TimeNetMissingParentError: If neither registry holds a complete payload source. + TimeFFormatError: If a manifest names the wrong release or a pin fails. """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers dest.check_writable(f"a copy of {reference}") if force and not isinstance(dest, LocalRegistry): @@ -88,16 +88,16 @@ def _copy_one( # ruff: ignore[too-many-arguments] force: bool, progress: ProgressCallback | None, ) -> None: - """Copy one release without its payload sources to ``dest``, or repair its local copy. + """Copy or repair one release without its payload sources in the destination registry. Args: reference: The release to copy. - manifest: The release's manifest from ``source``. + manifest: The release manifest from ``source``. source: The registry to copy from. dest: The registry to copy to. - staging: A temporary directory for a download from a non-local ``source``. - force: Fetch the release again, even if a local ``dest`` holds a complete copy. - progress: Receives the bytes of a local copy as they arrive. + staging: A temporary directory for downloads from a remote source. + force: Fetch the release again, even if ``dest`` has a complete local copy. + progress: Receives byte progress events when ``dest`` is local. """ # A local registry is a folder that the copy reads or writes directly. if isinstance(dest, LocalRegistry): @@ -117,15 +117,15 @@ def _copy_one( # ruff: ignore[too-many-arguments] def _bytes_of(reference: DatasetRef, manifest: Manifest, progress: ProgressCallback) -> Callable[[int], None]: - """Turn the byte counts of a release download into :class:`~timenet.progress.BytesMoved` events. + """Create a callback that reports cumulative bytes for a release. Args: - reference: The release. - manifest: The manifest of the release, whose file sizes give the total. - progress: Receives the events. + reference: The release to report. + manifest: The manifest whose file sizes define the total. + progress: The callback that receives ``BytesMoved`` events. Returns: - A function that takes the bytes of each file or chunk as it arrives, and sends the bytes so far. + A callback that accepts each file or chunk's byte count and reports the bytes moved so far. """ total = sum(part.size for part in manifest.files.all_files()) done = 0 diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py index e2dab398f..c28acec6d 100644 --- a/packages/timenet/src/timenet/registry/local.py +++ b/packages/timenet/src/timenet/registry/local.py @@ -327,14 +327,11 @@ def _latest_version(self, dataset_id: str) -> str | None: def is_same_registry(source: ReleaseSource, registry: BaseRegistry) -> bool: - """Return whether a source is a registry, or a local registry over the same directory. - - Two local registries over one directory hold the same releases, so a copy between them would - copy the directory onto itself. + """Check whether a source matches a registry. Args: - source: The source. - registry: The registry, for example the output of a build or the store of a client. + source: The source to compare. + registry: The registry to compare against. Returns: ``True`` for the same object or local registries over the same directory. From 7cf11576aca47105fdd96e1d4b7cd0fb6b209633 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Thu, 5 Nov 2026 10:37:56 +0100 Subject: [PATCH 4/5] test: trim the tests of the download into the store Drop tests that cannot fail or repeat other tests, and sharpen the rest. Part of TNET-157. --- .../timenet/tests/registry/test_remote.py | 12 ++-- packages/timenet/tests/test_cli.py | 7 +- packages/timenet/tests/test_client.py | 69 +++++-------------- 3 files changed, 27 insertions(+), 61 deletions(-) diff --git a/packages/timenet/tests/registry/test_remote.py b/packages/timenet/tests/registry/test_remote.py index bb7eb1c94..6a4337c35 100644 --- a/packages/timenet/tests/registry/test_remote.py +++ b/packages/timenet/tests/registry/test_remote.py @@ -204,16 +204,15 @@ def test_full_download_writes_identical_bytes(version_dir, tmp_path): assert (dest / "manifest.json").exists() -@pytest.mark.parametrize("version", [None, "1.0.0"]) -def test_a_download_into_the_store_resolves_a_remote_release_once(version_dir, tmp_path, version): +def test_a_download_into_the_store_reads_the_remote_manifest_for_the_plan_and_for_the_copy(version_dir, tmp_path): registry, requests = _remote(version_dir, tmp_path) directory, manifest = version_dir store = LocalRegistry(tmp_path / "store") - path = TimeNet(sources=[store, registry]).download(manifest.metadata.dataset_id, version) + path = TimeNet(sources=[store, registry]).download(manifest.metadata.dataset_id) assert (path / "manifest.json").read_bytes() == (directory / "manifest.json").read_bytes() - # One read of the manifest resolves the release, and one starts the copy. + # The plan reads the manifest once, and the copy reads it again before it writes. assert sum(request.url.path.endswith("/manifest") for request in requests) == 2 @@ -597,9 +596,8 @@ def _gated_release(origin: Path, page: str) -> tuple[str, DatasetRef]: @pytest.mark.parametrize("entry_point", ["open", "download"]) -@pytest.mark.parametrize("page", ["own", "ancestor"]) -def test_a_read_or_a_download_checks_access_before_it_reads_from_a_remote_registry(tmp_path, page, entry_point): - dataset_id, release = _gated_release(tmp_path / "origin", page) +def test_a_read_or_a_download_checks_access_before_it_reads_from_a_remote_registry(tmp_path, entry_point): + dataset_id, release = _gated_release(tmp_path / "origin", "ancestor") transport, requests = build_fake(next((tmp_path / "origin").rglob("manifest.json")).parent) remote = RemoteRegistry("http://api.local", transport=transport, cache_dir=tmp_path / "cache") client = TimeNet(sources=[LocalRegistry(tmp_path / "local"), remote]) diff --git a/packages/timenet/tests/test_cli.py b/packages/timenet/tests/test_cli.py index 5da0e8db3..991fcbede 100644 --- a/packages/timenet/tests/test_cli.py +++ b/packages/timenet/tests/test_cli.py @@ -8,6 +8,7 @@ import typer from typer.testing import CliRunner +from timenet import builders from timenet.cache import cached_datasets, raw_cache_size from timenet.cli.app import _enum_list, app from timenet.dataset import TimeFDataset @@ -153,6 +154,8 @@ def test_info_lists_the_access_page_of_each_release_that_sets_one(tmp_path): def test_download_command_copies_into_the_store_and_then_has_nothing_to_do(registry_root, tmp_path, monkeypatch): + # The default sources end with the installed connectors. Keep the real ones out of the resolution. + monkeypatch.setattr(builders, "entry_points", lambda group: []) # The last line names a long path, so the console must not wrap it. monkeypatch.setattr(_cli_module.ui, "rich", Console(stderr=True, width=1000)) monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home")) @@ -166,12 +169,12 @@ def test_download_command_copies_into_the_store_and_then_has_nothing_to_do(regis assert copied.exit_code == 0, copied.output assert copied.stdout == f"{target}\n" # only the directory, for a script to capture assert (target / "manifest.json").exists() - successes = [line for line in copied.stderr.splitlines() if line.startswith("✅")] + successes = [line for line in click.unstyle(copied.stderr).splitlines() if line.startswith("✅")] assert successes == ["✅ Copied 'timenet/hello-world' → 1.0.0"] assert again.exit_code == 0, again.output assert again.stdout == f"{target}\n" stored = tmp_path / "store" - assert again.stderr.splitlines() == [ + assert click.unstyle(again.stderr).splitlines() == [ f"✅ Nothing to do: 'timenet/hello-world' 1.0.0 is already in local registry {stored}" ] diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py index 474c70986..7f696b85d 100644 --- a/packages/timenet/tests/test_client.py +++ b/packages/timenet/tests/test_client.py @@ -140,15 +140,6 @@ def test_resampled_child_reads_and_downloads_without_its_build_only_parent(tmp_p assert reader.read().records[0].signals[0].to_arrow().equals(expected) -def test_a_damaged_release_names_no_download_without_a_writable_local_registry(tmp_path): - registry, _ = _child_of_the_demo(tmp_path / "registry") - (registry.root / "test/child/1.0.0/control.duckdb").write_bytes(b"cut short") - # download() copies into a writable local registry, so a client without one has no repair to offer. - with pytest.raises(TimeNetFileError) as caught: - TimeNet(sources=[LocalRegistry(registry.root, read_only=True)]).open("test/child") - assert caught.value.hint is None - - def test_a_damaged_release_outside_the_store_names_no_download(tmp_path): origin, _ = _child_of_the_demo(tmp_path / "origin") mirror = LocalRegistry(tmp_path / "mirror") @@ -178,10 +169,9 @@ def test_a_damaged_build_in_the_store_names_the_build_that_replaces_it(tmp_path, # No registry other than the local registry provides the release, so download() cannot replace it. hint = f"Build it again with `timenet-build build timenet/hello-world --out {store.root} --force`." - for entry_point in (client.open, client.load, client.download): - with pytest.raises(TimeNetFileError) as caught: - entry_point("timenet/hello-world@1.0.0") - assert caught.value.hint == hint + with pytest.raises(TimeNetFileError) as caught: + client.open("timenet/hello-world@1.0.0") + assert caught.value.hint == hint assert len(calls) == 1 # nothing builds over the damaged copy @@ -231,6 +221,7 @@ def test_the_default_list_follows_timenet_registry(tmp_path, monkeypatch, regist def test_timenet_registry_env_selects_a_local_registry(tmp_path, monkeypatch): + monkeypatch.setattr(builders_module, "entry_points", lambda group: []) monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home")) monkeypatch.setenv("TIMENET_REGISTRY", str(tmp_path / "home" / "registry")) dataset = make_dataset() @@ -332,15 +323,6 @@ def test_get_list_and_search_ask_the_first_registry_that_is_not_the_store(regist assert [metadata.dataset_id for metadata in alone.list()] == ["test/other"] -def test_download_copies_the_release_and_its_payload_sources_into_the_store(tmp_path): - origin, _ = _child_of_the_demo(tmp_path / "origin") - store = LocalRegistry(tmp_path / "store") - # The path of origin opens a writable local registry too, but only the first one gets the copies. - path = TimeNet(sources=[store, origin.root]).download("test/child") - assert path == tmp_path / "store/test/child/1.0.0" - assert store.exists("timenet/hello-world", "1.0.0") - - def test_download_needs_a_local_registry_that_is_not_read_only(registry_root): with pytest.raises(TimeFValidationError) as caught: TimeNet(sources=[LocalRegistry(registry_root, read_only=True)]).download("timenet/hello-world") @@ -438,8 +420,7 @@ def test_download_refuses_a_store_copy_with_other_files(registry_root, tmp_path, assert (directory / "manifest.json").read_text() == before -@pytest.mark.parametrize("case", ["force", "damaged"]) -def test_download_names_the_store_when_no_other_registry_provides_its_copy(tmp_path, monkeypatch, case): +def test_download_names_the_store_when_no_other_registry_provides_its_copy(tmp_path, monkeypatch): calls = [] install_backends(monkeypatch, _Builder(calls)) store = LocalRegistry(tmp_path / "store") @@ -447,17 +428,11 @@ def test_download_names_the_store_when_no_other_registry_provides_its_copy(tmp_p client = TimeNet(sources=[store, InstalledConnectors()]) # The store holds the release, so the error is about its copy, not about a build. - if case == "force": - message = r"^No registry other than the local registry provides timenet/hello-world@1\.0\.0, so download\(force=True\)" - with pytest.raises(TimeNetDatasetNotFoundError, match=message): - client.download("timenet/hello-world", force=True) - else: - (store.root / "timenet/hello-world/1.0.0/control.duckdb").write_bytes(b"cut short") - # The installed connector builds the release, so the hint names the build that replaces it. - hint = f"Build it again with `timenet-build build timenet/hello-world --out {store.root} --force`." - with pytest.raises(TimeNetFileError) as caught: - client.download("timenet/hello-world") - assert caught.value.hint == hint + message = ( + r"^No registry other than the local registry provides timenet/hello-world@1\.0\.0, so download\(force=True\)" + ) + with pytest.raises(TimeNetDatasetNotFoundError, match=message): + client.download("timenet/hello-world", force=True) assert calls == [] @@ -493,30 +468,23 @@ def test_download_counts_a_registry_over_the_directory_of_the_store_as_the_store assert {path: path.read_bytes() for path in directory.rglob("*") if path.is_file()} == before -def test_download_counts_a_case_variant_of_the_store_directory_as_the_store(tmp_path): - (tmp_path / "Probe").mkdir() - if not (tmp_path / "probe").exists(): - pytest.skip("the temporary folder tells upper and lower case apart") +def test_download_counts_a_symlink_to_the_store_directory_as_the_store(tmp_path): store = LocalRegistry(tmp_path / "store") store.put_dataset(make_dataset()) - client = TimeNet(sources=[store, LocalRegistry(tmp_path / "STORE")]) + (tmp_path / "link").symlink_to(store.root, target_is_directory=True) + client = TimeNet(sources=[store, LocalRegistry(tmp_path / "link")]) # The second registry reads the directory of the store, so it cannot copy the release again. with pytest.raises(TimeNetDatasetNotFoundError, match=r"^No registry other than the local registry provides "): client.download("timenet/hello-world", force=True) -@pytest.mark.parametrize("case", ["missing", "damaged"]) -def test_download_completes_the_payload_sources_of_a_complete_store_copy(tmp_path, case): +def test_download_completes_the_payload_sources_of_a_complete_store_copy(tmp_path): origin, reference = _child_of_the_demo(tmp_path / "origin") store = LocalRegistry(tmp_path / "store") client = TimeNet(sources=[store, origin]) client.download("test/child") - source = store.root / reference.dataset_id / str(reference.version) - if case == "missing": - shutil.rmtree(source) - else: - (source / "control.duckdb").write_bytes(b"cut short") + (store.root / reference.dataset_id / str(reference.version) / "control.duckdb").write_bytes(b"cut short") # The child is complete in the store, but the payload source that it reads is not. assert client.download("test/child") == store.root / "test/child/1.0.0" @@ -580,6 +548,8 @@ def test_download_repairs_a_damaged_store_copy(tmp_path): def test_download_returns_a_complete_store_copy_at_once(registry_root, tmp_path): client = TimeNet(sources=[LocalRegistry(tmp_path / "store"), str(registry_root)]) first = client.download("timenet/hello-world") + # The store copy is complete, so the next download needs no registry. + shutil.rmtree(registry_root) events: list[ProgressEvent] = [] assert client.download("timenet/hello-world", progress=events.append) == first assert events == [] # nothing is copied again @@ -778,11 +748,6 @@ def recording_open(client, *args, **kwargs): assert str(reader.metadata.dataset_version) == "1.0.0" -def test_version_ref_download_pins(versioned_registry, tmp_path): - client = TimeNet(sources=[LocalRegistry(tmp_path / "store"), str(versioned_registry)]) - assert client.download("timenet/hello-world@1.0.0") == tmp_path / "store/timenet/hello-world/1.0.0" - - def test_version_ref_missing_pin_raises(versioned_registry): client = TimeNet(sources=[str(versioned_registry)]) with pytest.raises(TimeNetDatasetNotFoundError): From b42d4af575f0b18ae832156bab8ec938cc8f2578 Mon Sep 17 00:00:00 2001 From: Adrian Blumer Date: Thu, 5 Nov 2026 11:44:19 +0100 Subject: [PATCH 5/5] refactor: mark the glue of the download into the store as a TODO The glue comments in cache.py now start with TODO(adrian), so a search finds each one that is left. A new TODO(adrian) comment in cli/app.py marks the glue that puts the --registry of timenet download after the store in the list of sources. Part of TNET-157. --- packages/timenet/src/timenet/cache.py | 13 ++++++++----- packages/timenet/src/timenet/cli/app.py | 2 ++ 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/packages/timenet/src/timenet/cache.py b/packages/timenet/src/timenet/cache.py index 2649ab31d..590697da3 100644 --- a/packages/timenet/src/timenet/cache.py +++ b/packages/timenet/src/timenet/cache.py @@ -4,9 +4,10 @@ download cache. All three sit under the home directory. """ -# Glue: the "download storage" of this module is the storage directory, settings().storage_dir. -# download() copies into the local registry now, so the storage directory holds only the copies that remote -# registries cache. A later change replaces this module and its view of the storage directory. +# TODO(adrian): Remove temporary glue which maps the "download storage" of this module onto the +# storage directory, settings().storage_dir, until a later change replaces this module and its +# view of the storage directory. download() copies into the local registry now, so the storage directory +# holds only the copies that remote registries cache. from dataclasses import dataclass import os @@ -53,7 +54,8 @@ def cached_datasets() -> list[CachedDataset]: The cached versions sorted by location, dataset ID, and version. """ config = settings() - # Glue: the storage directory, as the note at the top of this module says. + # TODO(adrian): Remove temporary glue which maps the download storage onto the storage + # directory until a later change replaces this module, as the note at the top says. found = _scan(config.registry_path, "registry") + _scan(config.storage_dir, "storage") return sorted(found, key=lambda dataset: (dataset.location, dataset.dataset_id, dataset.version)) @@ -80,7 +82,8 @@ def clear_cache(*, include_registry: bool) -> tuple[int, list[Path]]: A ``(bytes_freed, removed_dirs)`` pair with the removed directory paths. """ config = settings() - # Glue: the storage directory, as the note at the top of this module says. + # TODO(adrian): Remove temporary glue which maps the download storage onto the storage + # directory until a later change replaces this module, as the note at the top says. targets = [config.storage_dir, config.cache_dir] if include_registry: targets.append(config.registry_path) diff --git a/packages/timenet/src/timenet/cli/app.py b/packages/timenet/src/timenet/cli/app.py index 7e97f68ad..0134b4b01 100644 --- a/packages/timenet/src/timenet/cli/app.py +++ b/packages/timenet/src/timenet/cli/app.py @@ -28,6 +28,8 @@ "-r", help="Registry URL or path that this command reads (else $TIMENET_REGISTRY, then timenet://).", ) +# TODO(adrian): Remove temporary glue which maps --registry onto the list after the local registry, until the CLI +# takes the flags of the defaults. _download_registry_option = typer.Option( None, "--registry",