diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/__init__.py b/packages/timenet-connectors/src/timenet_connectors/builder/__init__.py index 38f17dcbf..6d0b3f21a 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/__init__.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/__init__.py @@ -1,6 +1 @@ -"""The ``timenet-build`` producer CLI.""" - -from timenet_connectors.builder.cli import app, main - - -__all__ = ["app", "main"] +"""Build datasets with connectors and the ``timenet-build`` CLI.""" diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/backend.py b/packages/timenet-connectors/src/timenet_connectors/builder/backend.py index ba5a35c2c..358458152 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/backend.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/backend.py @@ -1,4 +1,4 @@ -"""The build backend the SDK reaches through the ``timenet.builders`` entry point.""" +"""Read connector cards and build dataset releases.""" from collections.abc import Mapping from dataclasses import dataclass @@ -10,18 +10,17 @@ from timenet.errors import TimeNetDatasetNotFoundError, TimeNetVersionUnavailableError from timenet.manifest import PayloadSource from timenet.progress import BytesMoved, ProgressCallback, ShardWritten -from timenet.registry import BaseRegistry, LocalRegistry +from timenet.registry import LocalRegistry from timenet.types import DatasetMetadata from timenet.types.references import DatasetRef from timenet.writer.progress import ProgressStage, WriteProgressEvent -from timenet_connectors.builder.cli import build_one from timenet_connectors.builder.env import run_isolated_build -from timenet_connectors.discovery import connector_dir, has_connector, known_ids, resolve +from timenet_connectors.discovery import connector_dir, known_ids, resolve from timenet_connectors.download import DownloadProgress, progress_sink class ConnectorBuilder: - """Build a dataset with its connector, in an environment built from the connector's requirements.""" + """Build dataset releases with their connectors.""" def dataset_ids(self) -> list[str]: # ruff: ignore[no-self-use] (BuilderBackend protocol method) """List this package's dataset IDs without importing connectors. @@ -31,87 +30,27 @@ def dataset_ids(self) -> list[str]: # ruff: ignore[no-self-use] (BuilderBackend """ return known_ids() - def knows(self, dataset_id: str) -> bool: # ruff: ignore[no-self-use] (BuilderBackend protocol method) - """Report whether a connector package exists for the id, without importing it. - - Args: - dataset_id: The dataset id. - - Returns: - Whether a connector exists. - """ - return has_connector(dataset_id) - - def declared_version(self, dataset_id: str) -> str | None: - """Read the version the connector's card declares, without importing or building it. - - Args: - dataset_id: The dataset id. - - Returns: - The declared version string, or ``None`` if no connector exists for the dataset. - - Raises: - TimeNetInvalidCardError: If the card of the connector cannot be read or is not a mapping. - """ # ruff: ignore[docstring-extraneous-exception] - raised by card() - card = self.card(dataset_id) - return None if card is None else str(card.dataset_version) - def card(self, dataset_id: str) -> DatasetMetadata | None: # ruff: ignore[no-self-use] (protocol) - """Read the connector's ``dataset.yaml`` card, without importing or building the connector. + """Read a connector's dataset card without importing the connector. Args: - dataset_id: The dataset id. + dataset_id: The exact dataset ID declared in the card. Returns: - The validated card, or ``None`` if no connector exists for the dataset. + The validated card, or ``None`` if no connector card declares this ID. Raises: - TimeNetInvalidCardError: If the card cannot be read or is not a mapping. + TimeNetInvalidCardError: If reading the card fails or the card is not a mapping. """ # ruff: ignore[docstring-extraneous-exception] - raised by DatasetMetadata.from_yaml try: directory = connector_dir(dataset_id) except LookupError: return None - return DatasetMetadata.from_yaml(directory / "dataset.yaml") - - def build( # ruff: ignore[too-many-arguments, no-self-use] (protocol) - self, - dataset_id: str, - root: Path, - *, - force: bool = False, - build_parents: bool = False, - fetch_from: str | BaseRegistry | None = None, - pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, - ) -> Path: - """Build a dataset in an isolated environment. - - Set ``TIMENET_ISOLATION=off`` to run in the current interpreter. - - Args: - dataset_id: The dataset id. - root: The output registry directory. - force: Rebuild an existing version. - build_parents: Build missing parent releases from their installed connectors. - fetch_from: The registry, or its URI, to copy missing parent releases from. - pinned_by: Releases that read this dataset's files, with file pins the build must match. - - Returns: - The committed version directory. - """ - return Path( - build_one( - dataset_id, - root, - fetch_from=fetch_from, - build_parents=build_parents, - force=force, - pinned_by=pinned_by, - ) - ) + card = DatasetMetadata.from_yaml(directory / "dataset.yaml") + # The module path of a connector also accepts other spellings, such as "_" for "-". + return card if card.dataset_id == dataset_id else None - def build_prepared( + def build( self, release: DatasetRef, store: LocalRegistry, diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py index 9a244dc82..9207ee93b 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py @@ -6,21 +6,17 @@ ``CONNECTOR`` and a ``dataset.yaml`` card. You do not register it here. """ -from collections.abc import Mapping import json from pathlib import Path import typer -from timenet.builders import BuilderBackend, BuildOptions, declared, find_card, installed_builder +from timenet.builders import BuildOptions, declared from timenet.cli.runner import run_cli from timenet.cli.ui import console, progress_renderer from timenet.config import settings from timenet.coordinator.build import build_release -from timenet.coordinator.plan import Step, prepare_parents -from timenet.engine import publish_pipeline, run_pipeline from timenet.errors import ( - TimeFValidationError, TimeNetAccessError, TimeNetFileError, TimeNetMissingParentError, @@ -38,8 +34,8 @@ ) from timenet.registry._paths import release_dir # ruff: ignore[import-private-name] from timenet.types.references import DatasetRef, split_ref -from timenet_connectors.builder.env import run_isolated -from timenet_connectors.discovery import known_ids, resolve +from timenet_connectors.builder.backend import ConnectorBuilder +from timenet_connectors.discovery import known_ids app = typer.Typer(help="Build TimeNet datasets from connectors.", no_args_is_help=True) @@ -103,13 +99,6 @@ def build( "--isolation/--no-isolation", help="Run the build in an environment built from the connector's requirements.", ), - pinned_by: Path | None = typer.Option( - None, - "--pinned-by", - hidden=True, - help="A JSON file of the releases that read files of this release, with their pins. The parent " - "of an isolated build passes it.", - ), ) -> None: """Run a connector through the engine and write or publish its dataset. @@ -163,7 +152,6 @@ def build( fetch_from=None if fetch_from is None else open_registry(fetch_from), build_parents=build_parents, force=force, - pinned_by=None if pinned_by is None else _read_pins(pinned_by), options=BuildOptions( keep_cache=keep_cache, progress=progress, @@ -194,27 +182,25 @@ def _build_to( # ruff: ignore[too-many-arguments] fetch_from: BaseRegistry | None, build_parents: bool, force: bool, - pinned_by: Mapping[DatasetRef, PayloadSource] | None, options: BuildOptions, ) -> str: """Build a release locally and copy it to a remote output if needed. - The build reuses an existing release unless you set ``force``. + Reuse an existing release unless ``force`` is true. Args: dataset_id: The dataset ID. - target: The local output directory or remote registry. - fetch_from: The registry to copy missing parents from, or ``None`` to use the remote output if present. + target: The local directory or remote registry to write to. + fetch_from: The registry to copy missing parents from. ``None`` uses a remote output, if any. build_parents: Build missing parents with their installed connectors. - force: Rebuild the release even if it exists in the local registry. - pinned_by: The file records that dependent releases require this build to match. - options: The settings for each build. + force: Rebuild the release even if the local registry holds it. + options: Settings for each build. Returns: - The local version directory or remote version as a string. + The version directory for a local output, or the version string for a remote output. Raises: - BadParameter: If ``force`` is set and the version exists in the remote output. + BadParameter: If ``force`` is set and the remote output already holds the version. """ release = declared(dataset_id).release if isinstance(target, Path): @@ -241,7 +227,6 @@ def _build_to( # ruff: ignore[too-many-arguments] fetch_from=fetch_from, build_parents=build_parents, force=force, - pinned_by=pinned_by, open_access_only=remote is not None, options=options, ) @@ -263,203 +248,16 @@ def build_prepared_release_command( None, "--pinned-by", help="A JSON file of the releases that read files of this release, with their pins." ), ) -> None: - """Build a release in a local registry. + """Build one release into a local registry. - The registry must already contain all parent releases and their payload sources. - Status and progress go to stderr, and nothing goes to stdout. + The registry must already hold the release's parents and their payload sources. The command prints status + and progress to stderr. """ - # backend.py imports this module, so a module-level import here would be circular. - from timenet_connectors.builder.backend import ConnectorBuilder # ruff: ignore[import-outside-top-level] - reference = DatasetRef.parse(release) pins = {} if pinned_by is None else _read_pins(pinned_by) with progress_renderer() as progress: options = BuildOptions(keep_cache=keep_cache, progress=progress, quiet=console.quiet) - ConnectorBuilder().build_prepared(reference, LocalRegistry(store), force=force, pinned_by=pins, options=options) - - -def build_one( # ruff: ignore[too-many-arguments] - dataset_id: str, - output: str | Path, - *, - fetch_from: str | BaseRegistry | None = None, - build_parents: bool = False, - force: bool = False, - keep_cache: bool = False, - isolation: bool | None = None, - pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, -) -> str: - """Prepare parent releases and build a dataset. - - A missing parent comes from ``fetch_from`` or ``build_parents``, or the build fails before any download. If the - output has the version and ``force`` is off, the build only prepares missing releases that supply its data files. - - Args: - dataset_id: The dataset ID. - output: A local directory, or a ``timenet://``, ``http(s)://`` or ``s3://`` URL. - fetch_from: The registry, or its URI, to copy missing parent releases from. - build_parents: Build missing parents with installed connectors. - force: Rebuild the release even if it exists in the output. - keep_cache: Keep downloaded source files after the build. - isolation: Use a separate environment if true, this process if false, or ``TIMENET_ISOLATION`` if ``None``. - pinned_by: The file records that dependent releases require this build to match. - - Returns: - The local version directory or published version as a string. - """ - target = _resolve_target(str(output)) - registry = LocalRegistry(target) if isinstance(target, Path) else target - override = None if isolation is None else ("on" if isolation else "off") - isolated = settings(isolation=override).isolation == "on" - - def build_parent(step: Step) -> object: - found = installed_builder(step.release.dataset_id) - if found is not None and not _in_repo(found.backend): - return _build_with(found.backend, step, target) - # Each parent builds like the dataset itself. An isolated child makes the plan again. The - # child finds every parent in the output, so it only opens the parents. - return _build( - step.release.dataset_id, - target, - output, - isolated=isolated, - keep_cache=keep_cache, - pinned_by=step.pinned_by, - ) - - prepare_parents( - dataset_id, - registry, - build=build_parent, - fetch_from=open_registry(fetch_from) if isinstance(fetch_from, str) else fetch_from, - build_parents=build_parents, - force_build=force, - ) - committed = None if force else _committed(dataset_id, target, registry) - if committed is not None: - return committed - return _build( - dataset_id, - target, - output, - isolated=isolated, - force=force, - keep_cache=keep_cache, - pinned_by=pinned_by, - ) - - -def _in_repo(builder: BuilderBackend) -> bool: - """Report whether a parent that ``builder`` claims builds through :func:`_build`. - - :class:`~timenet_connectors.builder.backend.ConnectorBuilder` builds through :func:`_build`. A - subclass that does not override ``build()`` also builds through :func:`_build`. A subclass that - overrides ``build()`` builds through its own ``build()``. - - A backend that is not a ``ConnectorBuilder`` builds through its own ``build()``. This function - does not read ``build`` from the class of such a backend. So the backend can provide ``build`` - only on the instance, for example through ``__getattr__``. - - Returns: - Whether ``builder`` is a :class:`~timenet_connectors.builder.backend.ConnectorBuilder` and - its ``build()`` is the ``build()`` of ``ConnectorBuilder``. - """ - # backend.py imports this module, so this import runs only at call time. TNET-157 removes the cycle. - from timenet_connectors.builder.backend import ConnectorBuilder # ruff: ignore[import-outside-top-level] - - return isinstance(builder, ConnectorBuilder) and type(builder).build is ConnectorBuilder.build - - -def _build_with(builder: BuilderBackend, step: Step, target: WritableRegistry | Path) -> Path: - """Build a parent release with its file pins through the given backend. - - ``target`` must already hold the parents of ``step.release``. - - Returns: - The committed version directory. - - Raises: - TimeFValidationError: If ``target`` is not a local ``Path``. - """ - if not isinstance(target, Path): - raise TimeFValidationError( - f"parent release {step.release} builds through another builder backend, which writes to a " - "local directory; build it into a local output and publish it first" - ) - return builder.build(step.release.dataset_id, target, pinned_by=step.pinned_by or None) - - -def _committed(dataset_id: str, target: WritableRegistry | Path, registry: WritableRegistry) -> str | None: - """Return the result of a build whose version the output already holds. - - Returns: - The version directory for a local output, or the version for a remote output. ``None`` if - the output does not hold the version. - """ - card = find_card(dataset_id) - if card is None or not registry.exists(card.dataset_id, str(card.dataset_version)): - return None - version = str(card.dataset_version) - return str(target / card.dataset_id / version) if isinstance(target, Path) else version - - -def _build( # ruff: ignore[too-many-arguments] - dataset_id: str, - target: WritableRegistry | Path, - forwarded: str | Path, - *, - isolated: bool, - force: bool = False, - keep_cache: bool = False, - pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, -) -> str: - """Build a dataset in a local directory or remote registry. - - Args: - dataset_id: The dataset ID. - target: The output directory or registry when building in this interpreter. - forwarded: The output path or URI when building in a separate environment. - isolated: Build in a separate environment with the connector's required packages. - force: Rebuild the release even if it exists in the output. - keep_cache: Keep downloaded source files after the build. - pinned_by: The file records that dependent releases require this build to match. - - Returns: - The local version directory or published version as a string. - """ - pinned_by = pinned_by or None - if isolated: - # The child is this same CLI and inherits stderr, so it is the only narrator: it prints the - # status lines and the richer download and shard progress. This branch stays quiet and only - # relays what the child printed on stdout. Do not call resolve() here; it imports the connector - # module, which needs the dependencies the child is about to install. run_isolated forwards - # --out unchanged. So a remote target publishes from inside the environment, as a local - # target writes to a directory. - return run_isolated( - dataset_id, - forwarded, - force=force, - keep_cache=keep_cache, - quiet=console.quiet, - pinned_by=pinned_by, - ) - connector = resolve(dataset_id)() - if isinstance(target, Path): - version_dir = run_pipeline( - connector, - target, - force=force, - keep_cache=keep_cache, - pinned_by=pinned_by, - ) - return str(version_dir) - return publish_pipeline( - connector, - target, - force=force, - keep_cache=keep_cache, - pinned_by=pinned_by, - ) + ConnectorBuilder().build(reference, LocalRegistry(store), force=force, pinned_by=pins, options=options) def _read_pins(path: Path) -> dict[DatasetRef, PayloadSource]: diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/env.py b/packages/timenet-connectors/src/timenet_connectors/builder/env.py index e18936337..f8af26ef7 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/env.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/env.py @@ -18,7 +18,6 @@ import subprocess # ruff: ignore[suspicious-subprocess-import] - the command is a fixed argument vector import sys import tempfile -import threading from urllib.parse import urlparse from urllib.request import url2pathname @@ -99,49 +98,6 @@ def uv_command(spec: EnvSpec, argv: Sequence[str]) -> list[str]: return [*command, *argv] -def run_isolated( # ruff: ignore[too-many-arguments] - dataset_id: str, - out: str | Path, - *, - force: bool = False, - keep_cache: bool = False, - quiet: bool = False, - pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, -) -> str: - """Build a dataset in a separate environment. - - The build forwards stderr as it runs and includes its last lines in build errors. - - Args: - dataset_id: The dataset ID. - out: A local directory or a ``timenet://``, ``http(s)://``, or ``s3://`` URL. - force: Rebuild an existing version. - keep_cache: Keep the raw download cache after building. - quiet: Hide the build's status output. - pinned_by: File pins to preserve, grouped by dependent releases, or ``None``. - - Returns: - The local version directory or the published version for a remote target. - - Raises: - TimeNetBuildError: If the build fails or prints no result. - LookupError: If no connector exists for the dataset ID. - """ # ruff: ignore[docstring-extraneous-exception] (raised by _run_in_env, not directly here) - stdout = _run_in_env( - dataset_id, - ["build", dataset_id, "--out", str(out)], - force=force, - keep_cache=keep_cache, - quiet=quiet, - pinned_by=pinned_by, - subject=repr(dataset_id), - ) - lines = [line.strip() for line in stdout.splitlines() if line.strip()] - if not lines: - raise TimeNetBuildError(f"building {dataset_id!r} printed no version") - return lines[-1] - - def run_isolated_build( # ruff: ignore[too-many-arguments] release: DatasetRef, store_root: Path, @@ -151,80 +107,40 @@ def run_isolated_build( # ruff: ignore[too-many-arguments] quiet: bool, pinned_by: Mapping[DatasetRef, PayloadSource], ) -> None: - """Build a specific release into a local registry in its connector's environment. + """Build one release in its connector's environment. - The registry must already contain all parent releases and their payload sources. + The local registry must already contain the release's parents and their payload sources. Build errors include the last lines of stderr. Args: release: The exact dataset ID and version to build. store_root: The local registry directory to build into. - force: Rebuild even if the registry already contains the release. + force: Rebuild the release even if the registry already holds it. keep_cache: Keep the raw download cache after building. - quiet: Hide the build's status output. - pinned_by: Releases that depend on this release, with file pins that the build must preserve. + quiet: Hide build status messages. + pinned_by: File records the build must match, grouped by the release that pins them. Raises: TimeNetBuildError: If the build fails. LookupError: If no connector exists for the dataset ID. - """ # ruff: ignore[docstring-extraneous-exception] (raised by _run_in_env, not directly here) - _run_in_env( - release.dataset_id, - ["build-prepared-release", str(release), "--store", str(store_root)], - force=force, - keep_cache=keep_cache, - quiet=quiet, - pinned_by=pinned_by, - subject=str(release), - ) - - -def _run_in_env( # ruff: ignore[too-many-arguments] - dataset_id: str, - subcommand: Sequence[str], - *, - force: bool, - keep_cache: bool, - quiet: bool, - pinned_by: Mapping[DatasetRef, PayloadSource] | None, - subject: str, -) -> str: - """Run a ``timenet-build`` subcommand in a connector's environment. - - Args: - dataset_id: The dataset ID that selects the connector's environment. - subcommand: The subcommand and its arguments, before the build flags. - force: Add ``--force``. - keep_cache: Add ``--keep-cache``. - quiet: Add ``--quiet``. - pinned_by: The file pins to pass to the build, or ``None``. - subject: The name to include in build errors. - - Returns: - The command's stdout. - - Raises: - TimeNetBuildError: If the command fails. - LookupError: If no connector exists for the dataset ID. """ # ruff: ignore[docstring-extraneous-exception] (raised by env_spec, not directly here) - # --quiet belongs to timenet-build, not to the subcommand, so it goes first. - argv = ["timenet-build", *(["--quiet"] if quiet else []), *subcommand] - argv += [*(["--force"] if force else []), *(["--keep-cache"] if keep_cache else [])] - spec = env_spec(dataset_id) + # --quiet belongs to timenet-build, not to build-prepared-release, so it goes before the subcommand. + argv = ["timenet-build", *(["--quiet"] if quiet else []), "build-prepared-release", str(release)] + argv += ["--store", str(store_root), *(["--force"] if force else []), *(["--keep-cache"] if keep_cache else [])] + spec = env_spec(release.dataset_id) # TIMENET_ISOLATION=off is the recursion guard: the child is this same CLI. child_env = {**os.environ, "TIMENET_ISOLATION": "off"} with _pins_file(pinned_by) as pins: if pins is not None: argv += ["--pinned-by", str(pins)] command = uv_command(spec, argv) - stdout, stderr, returncode = _run_build(command, child_env) + stderr, returncode = _run_build(command, child_env) if returncode != 0: tail = "\n".join(stderr.splitlines()[-15:]).strip() detail = f"\n{tail}" if tail else " The child produced no error output." raise TimeNetBuildError( - f"building {subject} failed with exit code {returncode}. Command: {shlex.join(command)}.{detail}" + f"building {release} failed with exit code {returncode}. Command: {shlex.join(command)}.{detail}" ) - return stdout @contextmanager @@ -244,37 +160,29 @@ def _pins_file(pinned_by: Mapping[DatasetRef, PayloadSource] | None) -> Generato yield path -def _run_build(command: list[str], env: dict[str, str]) -> tuple[str, str, int]: +def _run_build(command: list[str], env: dict[str, str]) -> tuple[str, int]: """Run the build child, streaming its stderr live while also capturing it. - stdout is captured whole, because its last line is the version directory. stderr is echoed to - this process's stderr line by line, so a long download still shows progress, and captured too, so - a failure carries the child's own error. + stderr is echoed to this process's stderr line by line, so a long download still shows progress, + and captured too, so a failure carries the child's own error. Args: command: The full argument vector to run. env: The child's environment. Returns: - The child's captured stdout, its captured stderr, and its exit code. + The child's captured stderr and its exit code. """ process = subprocess.Popen( # ruff: ignore[subprocess-without-shell-equals-true] - command, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True + command, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True ) captured: list[str] = [] - - def tee_stderr() -> None: - for line in process.stderr or []: - sys.stderr.write(line) - sys.stderr.flush() - captured.append(line) - - pump = threading.Thread(target=tee_stderr) - pump.start() - stdout = process.stdout.read() if process.stdout is not None else "" + for line in process.stderr or []: + sys.stderr.write(line) + sys.stderr.flush() + captured.append(line) process.wait() - pump.join() - return stdout, "".join(captured), process.returncode + return "".join(captured), process.returncode def _base_interpreter() -> str: diff --git a/packages/timenet-connectors/src/timenet_connectors/discovery.py b/packages/timenet-connectors/src/timenet_connectors/discovery.py index ebdec3c4a..56803d3bb 100644 --- a/packages/timenet-connectors/src/timenet_connectors/discovery.py +++ b/packages/timenet-connectors/src/timenet_connectors/discovery.py @@ -52,22 +52,6 @@ def connector_dir(dataset_id: str) -> Path: raise LookupError(f"no connector for {dataset_id!r}; known: {', '.join(known_ids()) or '(none)'}") -def has_connector(dataset_id: str) -> bool: - """Check for a dataset's connector package and card without importing the connector. - - Args: - dataset_id: The dataset ID, as ``org/name``. - - Returns: - Whether a connector package with a ``dataset.yaml`` card exists. - """ - try: - connector_dir(dataset_id) - except LookupError: - return False - return True - - def requirements_for(dataset_id: str) -> Path | None: """Find a connector's requirements file. diff --git a/packages/timenet-connectors/tests/test_build_backend.py b/packages/timenet-connectors/tests/test_build_backend.py index 29e13c5b0..8c3baedb7 100644 --- a/packages/timenet-connectors/tests/test_build_backend.py +++ b/packages/timenet-connectors/tests/test_build_backend.py @@ -1,4 +1,3 @@ -from pathlib import Path import re import sys @@ -8,6 +7,7 @@ from timenet.errors import ( ConnectorCardWarning, TimeFValidationError, + TimeNetDatasetNotFoundError, TimeNetInvalidCardError, TimeNetVersionUnavailableError, ) @@ -20,7 +20,7 @@ from timenet.values_backends import ValuesBackend from timenet.writer.progress import ProgressStage, WriteProgressEvent from timenet_connectors import datasets as datasets_module -from timenet_connectors.builder import backend as backend_module, cli as cli_module +from timenet_connectors.builder import backend as backend_module from timenet_connectors.builder.backend import ConnectorBuilder from timenet_connectors.discovery import known_ids from timenet_connectors.download.progress import DownloadProgress, current_sink, progress_sink, report_progress @@ -30,7 +30,7 @@ def _unexpected_isolation(*args, **kwargs): - raise AssertionError(f"run_isolated should not have run: {args} {kwargs}") + raise AssertionError(f"run_isolated_build should not have run: {args} {kwargs}") def _unexpected_pipeline(*args, **kwargs): @@ -42,19 +42,6 @@ def _pins(): return {DatasetRef.parse("test/reader@1.0.0"): PayloadSource(release=HELLO, files=(pinned,))} -def test_knows_accepts_a_known_id(): - assert ConnectorBuilder().knows("timenet/hello-world") - - -def test_knows_rejects_an_unknown_id_without_importing_a_connector(): - leaf = "timenet_connectors.datasets.physionet.ecg_qa_cot" - sys.modules.pop(leaf, None) - - assert not ConnectorBuilder().knows("nope/nothing") - - assert leaf not in sys.modules - - def test_connector_builder_lists_its_dataset_ids_without_importing_a_connector(monkeypatch): leaf = "timenet_connectors.datasets.physionet.ecg_qa_cot" monkeypatch.delitem(sys.modules, leaf, raising=False) @@ -68,14 +55,27 @@ def test_connector_builder_lists_its_dataset_ids_without_importing_a_connector(m def test_a_card_that_cannot_be_read_raises_and_only_a_missing_connector_gives_none(tmp_path, monkeypatch): assert ConnectorBuilder().card("nope/nothing") is None - assert ConnectorBuilder().declared_version("nope/nothing") is None (tmp_path / "dataset.yaml").write_text("dataset_id: [unclosed\n") monkeypatch.setattr(backend_module, "connector_dir", lambda dataset_id: tmp_path) with pytest.raises(TimeNetInvalidCardError, match="could not read dataset card"): ConnectorBuilder().card("test/broken") - with pytest.raises(TimeNetInvalidCardError, match="could not read dataset card"): - ConnectorBuilder().declared_version("test/broken") + + +def test_the_card_answers_only_to_the_dataset_id_that_it_declares(tmp_path, monkeypatch): + # The module path of the connector accepts the spelling with "_", but a dataset ID must match its card. + assert ConnectorBuilder().card("timenet/hello-world") is not None + assert ConnectorBuilder().card("timenet/hello_world") is None + monkeypatch.setattr(backend_module, "run_isolated_build", _unexpected_isolation) + + with pytest.raises(TimeNetDatasetNotFoundError, match="timenet/hello_world"): + ConnectorBuilder().build( + DatasetRef.parse("timenet/hello_world@1.0.0"), + LocalRegistry(tmp_path), + force=False, + pinned_by={}, + options=BuildOptions(), + ) @pytest.mark.parametrize( @@ -101,60 +101,17 @@ def test_a_card_that_cannot_be_read_is_skipped_with_a_warning(tmp_path, monkeypa assert found.card.dataset_id == "timenet/hello-world" -def test_build_runs_in_an_isolated_environment_by_default(tmp_path, monkeypatch): - monkeypatch.delenv("TIMENET_ISOLATION", raising=False) - calls = [] - - def fake_isolated(dataset_id, root, *, force=False, **_): - calls.append((dataset_id, root, force)) - return Path(root) / "1.0.0" - - monkeypatch.setattr(cli_module, "run_isolated", fake_isolated) - - version_dir = ConnectorBuilder().build("timenet/hello-world", tmp_path, force=True) - - assert version_dir == tmp_path / "1.0.0" - assert calls == [("timenet/hello-world", tmp_path, True)] - - -def test_build_hands_its_pins_to_the_build(tmp_path, monkeypatch): - monkeypatch.delenv("TIMENET_ISOLATION", raising=False) - received = [] - - def fake_isolated(dataset_id, root, *, pinned_by=None, **_): - received.append(pinned_by) - return str(Path(root) / "1.0.0") - - monkeypatch.setattr(cli_module, "run_isolated", fake_isolated) - pinned = FilePart(path="control.duckdb", checksum="sha256:" + "0" * 64, size=1) - pins = { - DatasetRef.parse("test/reader@1.0.0"): PayloadSource( - release=DatasetRef.parse("timenet/hello-world@1.0.0"), files=(pinned,) - ) - } - - ConnectorBuilder().build("timenet/hello-world", tmp_path, pinned_by=pins) - - assert received == [pins] - - -def test_build_runs_in_process_when_isolation_is_off(tmp_path, monkeypatch): - monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home")) - monkeypatch.setenv("TIMENET_ISOLATION", "off") - monkeypatch.setattr(cli_module, "run_isolated", _unexpected_isolation) - - version_dir = ConnectorBuilder().build("timenet/hello-world", tmp_path / "registry") - - assert (version_dir / "manifest.json").is_file() - - def test_build_in_process_round_trips_the_connector_default_zarr_backend(tmp_path, monkeypatch): pytest.importorskip("zarr") - connector_cls = cli_module.resolve("timenet/hello-world") + connector_cls = backend_module.resolve("timenet/hello-world") monkeypatch.setattr(connector_cls, "values_backend", ValuesBackend.ZARR) monkeypatch.setenv("TIMENET_ISOLATION", "off") monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home")) - version_dir = ConnectorBuilder().build("timenet/hello-world", tmp_path / "registry") + store = LocalRegistry(tmp_path / "registry") + + ConnectorBuilder().build(HELLO, store, force=False, pinned_by={}, options=BuildOptions()) + + version_dir = tmp_path / "registry/timenet/hello-world/1.0.0" assert Manifest.model_validate_json((version_dir / "manifest.json").read_text()).values_backend == "zarr" with TimeFReader(open_local_version(version_dir)) as reader: dataset = reader.read() @@ -164,7 +121,7 @@ def test_build_in_process_round_trips_the_connector_default_zarr_backend(tmp_pat # Only BuildOptions.quiet silences the child. An SDK build without a progress callback still prints. @pytest.mark.parametrize("quiet", [False, True]) -def test_build_prepared_runs_an_isolated_child_with_the_release_and_the_pins(tmp_path, monkeypatch, quiet): +def test_build_runs_an_isolated_child_with_the_release_and_the_pins(tmp_path, monkeypatch, quiet): monkeypatch.delenv("TIMENET_ISOLATION", raising=False) monkeypatch.setattr(backend_module, "run_pipeline", _unexpected_pipeline) calls = [] @@ -172,7 +129,7 @@ def test_build_prepared_runs_an_isolated_child_with_the_release_and_the_pins(tmp backend_module, "run_isolated_build", lambda release, root, **options: calls.append((release, root, options)) ) - ConnectorBuilder().build_prepared( + ConnectorBuilder().build( HELLO, LocalRegistry(tmp_path), force=True, @@ -183,23 +140,21 @@ def test_build_prepared_runs_an_isolated_child_with_the_release_and_the_pins(tmp assert calls == [(HELLO, tmp_path, {"force": True, "keep_cache": True, "quiet": quiet, "pinned_by": _pins()})] -def test_build_prepared_rejects_another_version(tmp_path, monkeypatch): +def test_build_rejects_another_version(tmp_path, monkeypatch): monkeypatch.setenv("TIMENET_ISOLATION", "on") monkeypatch.setattr(backend_module, "run_isolated_build", _unexpected_isolation) monkeypatch.setattr(backend_module, "run_pipeline", _unexpected_pipeline) release = DatasetRef.parse("timenet/hello-world@9.9.9") with pytest.raises(TimeNetVersionUnavailableError) as caught: - ConnectorBuilder().build_prepared( - release, LocalRegistry(tmp_path), force=False, pinned_by={}, options=BuildOptions() - ) + ConnectorBuilder().build(release, LocalRegistry(tmp_path), force=False, pinned_by={}, options=BuildOptions()) # It is a TimeFValidationError too, so a handler of the old error still catches it. assert isinstance(caught.value, TimeFValidationError) assert (caught.value.release, str(caught.value.declared)) == (release, "1.0.0") -def test_build_prepared_reports_downloads_and_shards(tmp_path, monkeypatch): +def test_build_reports_downloads_and_shards(tmp_path, monkeypatch): class Connector: pass @@ -217,7 +172,7 @@ def pipeline(connector, root, *, progress_cb, **options): monkeypatch.setattr(backend_module, "run_pipeline", pipeline) events = [] - ConnectorBuilder().build_prepared( + ConnectorBuilder().build( HELLO, LocalRegistry(tmp_path), force=True, @@ -230,7 +185,7 @@ def pipeline(connector, root, *, progress_cb, **options): assert received == expected -def test_build_prepared_without_progress_installs_no_sink(tmp_path, monkeypatch): +def test_build_without_progress_installs_no_sink(tmp_path, monkeypatch): monkeypatch.setenv("TIMENET_ISOLATION", "off") seen = {} @@ -242,7 +197,7 @@ def pipeline(connector, root, *, progress_cb, **options): # An outer sink must not reach the build either: the build installs its own, empty one. with progress_sink(print): - ConnectorBuilder().build_prepared( + ConnectorBuilder().build( HELLO, LocalRegistry(tmp_path), force=False, pinned_by={}, options=BuildOptions(progress=None) ) diff --git a/packages/timenet-connectors/tests/test_build_cli.py b/packages/timenet-connectors/tests/test_build_cli.py index d2568224e..308f05928 100644 --- a/packages/timenet-connectors/tests/test_build_cli.py +++ b/packages/timenet-connectors/tests/test_build_cli.py @@ -288,7 +288,7 @@ def test_the_build_prepared_release_command_passes_the_pins_and_the_options(monk calls = [] monkeypatch.setattr( ConnectorBuilder, - "build_prepared", + "build", lambda self, release, store, **kwargs: calls.append((release, store, kwargs)), ) monkeypatch.setattr(cli_module.console, "quiet", False) diff --git a/packages/timenet-connectors/tests/test_build_env.py b/packages/timenet-connectors/tests/test_build_env.py index 5c0061440..ff6c91594 100644 --- a/packages/timenet-connectors/tests/test_build_env.py +++ b/packages/timenet-connectors/tests/test_build_env.py @@ -11,7 +11,10 @@ from timenet.manifest.files import FilePart from timenet.types.references import DatasetRef from timenet_connectors.builder import env as env_module -from timenet_connectors.builder.env import EnvSpec, env_spec, run_isolated, run_isolated_build, uv_command +from timenet_connectors.builder.env import EnvSpec, env_spec, run_isolated_build, uv_command + + +HELLO = DatasetRef.parse("timenet/hello-world@1.0.0") def _spec(requirements=None, base=("--with", "timenet==0.1.0")): @@ -117,101 +120,30 @@ def test_isolated_environment_installs_backend_dependencies(monkeypatch, tmp_pat def run(command, env): captured["command"] = command - return str(tmp_path), "", 0 + return "", 0 monkeypatch.setattr(env_module, "_run_build", run) - run_isolated("timenet/hello-world", tmp_path) + run_isolated_build(HELLO, tmp_path, force=False, keep_cache=False, quiet=False, pinned_by={}) dependencies = captured["command"][: captured["command"].index("timenet-build")] assert "timenet[zarr]==9.9.9" in dependencies -def test_run_isolated_disables_isolation_in_the_child(monkeypatch, tmp_path): - captured = {} - - def _fake(command, env): - captured["env"] = env - return f"{tmp_path}/timenet/hello-world/1.0.0\n", "", 0 - - monkeypatch.setattr(env_module, "_run_build", _fake) - run_isolated("timenet/hello-world", tmp_path) - - assert captured["env"]["TIMENET_ISOLATION"] == "off" - - -def test_run_isolated_returns_the_version_directory(monkeypatch, tmp_path): - stdout = f"noise\n{tmp_path}/timenet/hello-world/1.0.0\n" - monkeypatch.setattr(env_module, "_run_build", lambda command, env: (stdout, "", 0)) - - assert run_isolated("timenet/hello-world", tmp_path) == f"{tmp_path}/timenet/hello-world/1.0.0" - - -def test_run_isolated_raises_on_a_failed_child(monkeypatch, tmp_path): - monkeypatch.setattr(env_module, "_run_build", lambda command, env: ("", "", 2)) - - with pytest.raises(TimeNetBuildError, match="exit code 2"): - run_isolated("timenet/hello-world", tmp_path) - - -def test_run_isolated_failure_message_includes_the_child_stderr(monkeypatch, tmp_path): - monkeypatch.setattr(env_module, "_run_build", lambda command, env: ("", "boom: the disk is full", 1)) - - with pytest.raises(TimeNetBuildError, match="the disk is full"): - run_isolated("timenet/hello-world", tmp_path) - - -def test_run_isolated_raises_when_the_child_prints_no_directory(monkeypatch, tmp_path): - monkeypatch.setattr(env_module, "_run_build", lambda command, env: ("\n", "", 0)) - - with pytest.raises(TimeNetBuildError, match="no version"): - run_isolated("timenet/hello-world", tmp_path) - - -def test_run_isolated_forwards_the_build_flags(monkeypatch, tmp_path): - captured = {} - - def _fake(command, env): - captured["command"] = command - return f"{tmp_path}/x\n", "", 0 - - monkeypatch.setattr(env_module, "_run_build", _fake) - run_isolated("timenet/hello-world", tmp_path, force=True, keep_cache=True) - - assert "--force" in captured["command"] - assert "--keep-cache" in captured["command"] - - -def test_run_isolated_forwards_quiet_ahead_of_the_subcommand(monkeypatch, tmp_path): - captured = {} - - def _fake(command, env): - captured["command"] = command - return f"{tmp_path}/x\n", "", 0 - - monkeypatch.setattr(env_module, "_run_build", _fake) - run_isolated("timenet/hello-world", tmp_path, quiet=True) - - command = captured["command"] - # --quiet is a root option on timenet-build, so build never sees it. - assert command.index("--quiet") == command.index("build") - 1 - - -def test_run_isolated_omits_quiet_when_the_parent_is_not_quiet(monkeypatch, tmp_path): +def test_run_isolated_build_omits_quiet_when_the_parent_is_not_quiet(monkeypatch, tmp_path): captured = {} def _fake(command, env): captured["command"] = command - return f"{tmp_path}/x\n", "", 0 + return "", 0 monkeypatch.setattr(env_module, "_run_build", _fake) - run_isolated("timenet/hello-world", tmp_path) + run_isolated_build(HELLO, tmp_path, force=False, keep_cache=False, quiet=False, pinned_by={}) assert "--quiet" not in captured["command"] -def test_run_build_captures_stdout_and_stderr_and_the_exit_code(): - command = [sys.executable, "-c", "import sys; print('out'); print('an error', file=sys.stderr); sys.exit(3)"] - stdout, stderr, code = env_module._run_build(command, dict(os.environ)) - assert stdout.strip() == "out" +def test_run_build_captures_stderr_and_the_exit_code(): + command = [sys.executable, "-c", "import sys; print('an error', file=sys.stderr); sys.exit(3)"] + stderr, code = env_module._run_build(command, dict(os.environ)) assert "an error" in stderr assert code == 3 @@ -242,7 +174,7 @@ def run(command, env): # The pins file exists only while the child runs. pins = json.loads(Path(command[command.index("--pinned-by") + 1]).read_text(encoding="utf-8")) captured.update(argv=command[command.index("timenet-build") :], env=env, pins=pins) - return "", "", 0 + return "", 0 monkeypatch.setattr(env_module, "_run_build", run) pins = {DatasetRef.parse("test/reader@1.0.0"): pinned} @@ -255,3 +187,10 @@ def run(command, env): assert {"--force", "--keep-cache"} <= set(argv) assert captured["pins"] == {"test/reader@1.0.0": pinned.model_dump(mode="json")} assert captured["env"]["TIMENET_ISOLATION"] == "off" + + +def test_run_isolated_build_raises_with_the_tail_of_the_child_stderr(monkeypatch, tmp_path): + monkeypatch.setattr(env_module, "_run_build", lambda command, env: ("boom: the disk is full", 1)) + + with pytest.raises(TimeNetBuildError, match=r"timenet/hello-world@1\.0\.0.*exit code 1(.|\n)*the disk is full"): + run_isolated_build(HELLO, tmp_path, force=False, keep_cache=False, quiet=False, pinned_by={}) diff --git a/packages/timenet-connectors/tests/test_build_plan.py b/packages/timenet-connectors/tests/test_build_plan.py index 4df986233..82b3b35ee 100644 --- a/packages/timenet-connectors/tests/test_build_plan.py +++ b/packages/timenet-connectors/tests/test_build_plan.py @@ -8,7 +8,8 @@ from timenet import builders from timenet.connectors import BaseConnector -from timenet.coordinator.plan import Step, plan_parents, prepare_parents +from timenet.coordinator.build import build_release +from timenet.coordinator.plan import Step, plan_parents from timenet.dataset import Record, RegularAxis, Signal, Source, TimeFDataset from timenet.engine import open_parents, run_pipeline from timenet.errors import ( @@ -21,7 +22,7 @@ TimeNetMissingParentError, ) from timenet.reader import TimeFReader -from timenet.registry import LocalRegistry, S3Registry, WritableRegistry +from timenet.registry import LocalRegistry, WritableRegistry from timenet.registry._paths import release_dir from timenet.testing import install_backends, make_dataset from timenet.types import Access, TimeSeriesSpec, ureg @@ -52,17 +53,11 @@ def __init__(self, cards): def dataset_ids(self): return list(self.cards) - def knows(self, dataset_id): - return dataset_id in self.cards - - def declared_version(self, dataset_id): - return str(self.cards[dataset_id].dataset_version) - def card(self, dataset_id): return self.cards.get(dataset_id) - def build(self, dataset_id, root, *, force=False): - raise AssertionError(f"the plan does not build {dataset_id}") + def build(self, release, store, *, force, pinned_by, options): + raise AssertionError(f"the plan does not build {release}") class _InRepoCards(ConnectorBuilder): @@ -74,9 +69,6 @@ def __init__(self, cards): def dataset_ids(self): return list(self.cards) - def knows(self, dataset_id): - return dataset_id in self.cards - def card(self, dataset_id): return self.cards.get(dataset_id) @@ -92,10 +84,6 @@ def _cards(**parents: dict[str, DatasetRef]) -> dict: return cards -def _no_build(step: Step) -> None: - pytest.fail(f"the plan built {step.release}") - - def _install_cards(monkeypatch, **parents: dict[str, DatasetRef]) -> dict: """Register this package's builder backend, serving the cards of ``_cards(**parents)``. @@ -147,21 +135,16 @@ def test_plan_uses_fetches_or_builds_each_parent_once_before_its_children(tmp_pa TimeFValidationError, "cycle: test/child@1.0.0 -> test/parent@1.0.0 -> test/child@1.0.0", ), - ({"child": {"parent": PARENT}}, {"remote_output": True}, TimeFValidationError, "local output"), ], - ids=["missing", "other-version", "cycle", "fetch-into-remote"], + ids=["missing", "other-version", "cycle"], ) # ruff: ignore[too-many-arguments, too-many-positional-arguments] def test_plan_rejects_unobtainable_parents_before_anything_runs(tmp_path, monkeypatch, cards, options, error, match): _install_cards(monkeypatch, **cards) output = LocalRegistry(tmp_path / "output") - fetch_from = None - if options.pop("remote_output", False): - output = S3Registry("s3://bucket/registry", cache_dir=tmp_path / "cache") - fetch_from = LocalRegistry(tmp_path / "origin") with pytest.raises(error, match=match): - plan_parents(CHILD.dataset_id, output, fetch_from=fetch_from, **options) + plan_parents(CHILD.dataset_id, output, **options) _SPEC = TimeSeriesSpec(spec_type="sine", name="Sine", unit_value=ureg.dimensionless) @@ -200,7 +183,7 @@ def _declared_without_its_sources_source(registry: LocalRegistry) -> None: @pytest.mark.parametrize("where", ["output", "origin", "source-in-origin"]) -def test_a_declared_parent_needs_only_its_direct_payload_sources(tmp_path, monkeypatch, where): +def test_a_declared_parent_needs_only_its_direct_payload_sources(tmp_path, monkeypatch, child_builds, where): _install_cards(monkeypatch, child={"declared": DECLARED}) output, origin = LocalRegistry(tmp_path / "output"), LocalRegistry(tmp_path / "origin") _declared_without_its_sources_source(origin if where == "origin" else output) @@ -210,13 +193,24 @@ def test_a_declared_parent_needs_only_its_direct_payload_sources(tmp_path, monke steps = plan_parents(CHILD.dataset_id, output, fetch_from=origin) - # DECLARED opens with SOURCE, its direct payload source, which a fetch of DECLARED brings; GRAND, - # SOURCE's own source, is not needed. + # DECLARED opens with SOURCE, its direct payload source, which a copy of DECLARED brings. So the + # plan does not name GRAND, SOURCE's own source. assert steps == ([] if where == "output" else [Step(DECLARED, "copy")]) - prepare_parents(CHILD.dataset_id, output, build=_no_build, fetch_from=origin) + if where == "source-in-origin": + # The output holds DECLARED, so the run copies SOURCE with its own payload sources. That + # includes GRAND, which no registry holds, so the copy fails before it copies anything. + with pytest.raises( + TimeNetMissingParentError, match=r"^payload source timenet/hello-world@1\.0\.0 of test/source" + ): + build_release(CHILD.dataset_id, store=output, fetch_from=origin) + assert child_builds == [] + return + build_release(CHILD.dataset_id, store=output, fetch_from=origin) for reference in (DECLARED, SOURCE): assert output.exists(reference.dataset_id, str(reference.version)) assert not output.exists(GRAND.dataset_id, str(GRAND.version)) + # Only the requested release builds. + assert child_builds == [(CHILD.dataset_id, False)] def test_a_payload_source_that_is_later_a_declared_parent_brings_its_own_payload_sources(tmp_path, monkeypatch): @@ -304,17 +298,27 @@ def test_a_pinned_source_is_rebuilt_in_staging_and_published_only_if_it_matches( steps = plan_parents(CHILD.dataset_id, output, build_parents=True) assert steps == [Step(GRAND, "build", pinned_by={PARENT: pins})] + built = [] + + def child(release, store_root, *, pinned_by, **_): + built.append(release) + if release == GRAND: + run_pipeline(_Grand(), store_root, pinned_by=pinned_by) + else: + _store_from_card(release.dataset_id, store_root, force=False) - def build(step): - run_pipeline(_Grand(), output.root, pinned_by=step.pinned_by) + monkeypatch.delenv("TIMENET_ISOLATION", raising=False) + monkeypatch.setattr(backend_module, "run_isolated_build", child) if backend == ValuesBackend.PARQUET: - prepare_parents(CHILD.dataset_id, output, build=build, build_parents=True) + build_release(CHILD.dataset_id, store=output, build_parents=True) assert output.exists(GRAND.dataset_id, str(GRAND.version)) + assert built == [GRAND, CHILD] return # A Zarr rebuild writes other files than the Parquet files PARENT reads, so nothing is published. with pytest.raises(TimeFFormatError, match=r"Nothing was published\. Fetch the exact release .* or rebuild"): - prepare_parents(CHILD.dataset_id, output, build=build, build_parents=True) + build_release(CHILD.dataset_id, store=output, build_parents=True) + assert built == [GRAND] assert not output.exists(GRAND.dataset_id, str(GRAND.version)) assert not list(release_dir(output.root, GRAND).parent.glob("*")) @@ -356,15 +360,15 @@ def test_a_built_release_needs_only_its_payload_sources(tmp_path, monkeypatch): origin = LocalRegistry(tmp_path / "origin") origin.put_dataset(make_dataset()) assert plan_parents(PARENT.dataset_id, output, fetch_from=origin) == [Step(PARENT, "copy")] - prepare_parents(PARENT.dataset_id, output, build=_no_build, fetch_from=origin) - assert output.exists(GRAND.dataset_id, str(GRAND.version)) -def test_a_damaged_parent_is_fetched_again_but_never_built_over(tmp_path, monkeypatch): - _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) +def test_a_damaged_parent_is_fetched_again_but_never_built_over(tmp_path, monkeypatch, child_builds): + cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) output, origin = LocalRegistry(tmp_path / "output"), LocalRegistry(tmp_path / "origin") + parent = make_dataset() + parent._metadata = cards[PARENT.dataset_id] for registry in (output, origin): - registry.put_dataset(TimeFDataset(metadata=_cards(parent={})[PARENT.dataset_id])) + registry.put_dataset(parent) (release_dir(output.root, PARENT) / "control.duckdb").write_bytes(b"cut short") with pytest.raises(TimeNetFileError, match=r"'control\.duckdb' has 9 bytes, expected [\d,]+ bytes\.$") as caught: @@ -372,8 +376,9 @@ def test_a_damaged_parent_is_fetched_again_but_never_built_over(tmp_path, monkey assert caught.value.release == PARENT assert plan_parents(CHILD.dataset_id, output, fetch_from=origin) == [Step(PARENT, "copy")] - prepare_parents(CHILD.dataset_id, output, build=_no_build, fetch_from=origin) + build_release(CHILD.dataset_id, store=output, fetch_from=origin) assert plan_parents(CHILD.dataset_id, output) == [] + assert child_builds == [(CHILD.dataset_id, False)] class _Hosted(WritableRegistry): @@ -409,33 +414,33 @@ def put_release(self, from_directory): @pytest.mark.parametrize("gated", ["child-card", "held-parent", "parent-card"]) -def test_a_remote_output_refuses_a_plan_that_visits_a_release_that_is_not_open(tmp_path, monkeypatch, gated): +def test_open_access_only_refuses_a_plan_that_visits_a_release_that_is_not_open(tmp_path, monkeypatch, gated): cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) gated_ref = CHILD if gated == "child-card" else PARENT cards[gated_ref.dataset_id] = cards[gated_ref.dataset_id].model_copy(update=_GATED) - output = _Hosted(tmp_path / "hosted") + store = LocalRegistry(tmp_path / "store") if gated == "held-parent": - output.put_dataset(TimeFDataset(metadata=cards[PARENT.dataset_id])) + store.put_dataset(TimeFDataset(metadata=cards[PARENT.dataset_id])) with pytest.raises(TimeNetAccessError) as caught: - plan_parents(CHILD.dataset_id, output, build_parents=True) + plan_parents(CHILD.dataset_id, store, build_parents=True, open_access_only=True) error = caught.value requirement = AccessRequirement(gated_ref, Access.CREDENTIALED, _GATED["access_url"]) assert (error.dataset, error.access, error.requirements) == (gated_ref, Access.CREDENTIALED, (requirement,)) -def test_a_remote_output_names_the_access_page_of_a_held_parent_from_its_lineage(tmp_path, monkeypatch): +def test_open_access_only_names_the_access_page_of_a_held_parent_from_its_lineage(tmp_path, monkeypatch): cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) - output = _Hosted(tmp_path / "hosted") + store = LocalRegistry(tmp_path / "store") # The card of the held parent stays open. Its lineage holds GRAND, which is credentialed. held = TimeFDataset(metadata=cards[PARENT.dataset_id]) grand = make_dataset().metadata.model_copy(update=_GATED) held._import_state.lineage[str(GRAND)] = ParentDatasetCard(metadata=grand) - output.put_dataset(held) + store.put_dataset(held) with pytest.raises(TimeNetAccessError) as caught: - plan_parents(CHILD.dataset_id, output, build_parents=True) + plan_parents(CHILD.dataset_id, store, build_parents=True, open_access_only=True) assert (caught.value.dataset, caught.value.access) == (PARENT, Access.CREDENTIALED) assert caught.value.requirements == (AccessRequirement(GRAND, Access.CREDENTIALED, _GATED["access_url"]),) @@ -458,16 +463,6 @@ def _store_from_card(dataset_id: str, root: Path, *, force: bool) -> Path: return registry.root / dataset_id / str(metadata.dataset_version) -def _isolated_builds(calls: list) -> object: - """Return a stand-in for ``run_isolated`` that stores each dataset from its card and records the call.""" - - def run_isolated(dataset_id, root, *, force=False, **_): - calls.append((dataset_id, force)) - return str(_store_from_card(dataset_id, root, force=force)) - - return run_isolated - - def _prepared_builds(calls: list) -> object: """Return a stand-in for ``run_isolated_build`` that stores each release from its card and records the call.""" @@ -478,6 +473,19 @@ def run_isolated_build(release, store_root, *, force=False, **_): return run_isolated_build +@pytest.fixture +def child_builds(monkeypatch) -> list: + """Stand in for each isolated build with :func:`_prepared_builds`. + + Returns: + The list that receives one ``(dataset_id, force)`` entry for each build. + """ + monkeypatch.delenv("TIMENET_ISOLATION", raising=False) + calls = [] + monkeypatch.setattr(backend_module, "run_isolated_build", _prepared_builds(calls)) + return calls + + def test_build_prepares_parents_before_the_isolated_child(tmp_path, monkeypatch): cards = _install_cards(monkeypatch, child={"parent": PARENT, "grand": GRAND}, parent={}) monkeypatch.setattr(cli_module, "known_ids", lambda: list(cards)) @@ -500,43 +508,6 @@ def test_build_prepares_parents_before_the_isolated_child(tmp_path, monkeypatch) assert result.stdout.strip() == str(out / CHILD.dataset_id / str(CHILD.version)) -def test_a_built_version_starts_no_isolated_child(tmp_path, monkeypatch): - monkeypatch.delenv("TIMENET_ISOLATION", raising=False) - monkeypatch.setattr(env_module.subprocess, "Popen", lambda *args, **kwargs: pytest.fail("started a subprocess")) - out = LocalRegistry(tmp_path / "out") - out.put_dataset(make_dataset()) - - assert cli_module.build_one(GRAND.dataset_id, out.root) == str(release_dir(out.root, GRAND)) - - -@pytest.mark.parametrize("entry", ["api", "backend"]) -def test_every_build_entry_point_prepares_parents(tmp_path, monkeypatch, entry): - cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) - monkeypatch.delenv("TIMENET_ISOLATION", raising=False) - calls = [] - monkeypatch.setattr(cli_module, "run_isolated", _isolated_builds(calls)) - monkeypatch.setattr(backend_module, "run_isolated_build", _prepared_builds(calls)) - build = timenet_connectors.build if entry == "api" else ConnectorBuilder().build - out = tmp_path / "out" - - with pytest.raises( - TimeNetMissingParentError, match=r"parent release test/parent@1\.0\.0 of test/child@1\.0\.0" - ) as caught: - build(CHILD.dataset_id, **{"out" if entry == "api" else "root": out}) - assert calls == [] - if entry == "api": - assert str(caught.value).endswith(". Copy it with fetch_from= or build it with build_parents=True.") - - # A registry instance to fetch from, as the SDK layers accept one. - origin = LocalRegistry(tmp_path / "origin") - parent = make_dataset() - parent._metadata = cards[PARENT.dataset_id] - origin.put_dataset(parent) - build(CHILD.dataset_id, fetch_from=origin, **{"out" if entry == "api" else "root": out}) - assert calls == [(CHILD.dataset_id, False)] - assert LocalRegistry(out).exists(PARENT.dataset_id, str(PARENT.version)) - - @pytest.mark.parametrize("obtain", ["fetch_from", "build_parents"]) def test_load_obtains_a_missing_parent_before_it_builds_and_reads(tmp_path, monkeypatch, obtain): cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) @@ -550,8 +521,9 @@ def test_load_obtains_a_missing_parent_before_it_builds_and_reads(tmp_path, monk parent._metadata = cards[PARENT.dataset_id] origin.put_dataset(parent) - with pytest.raises(TimeNetMissingParentError, match=r"^parent release test/parent@1\.0\.0 of test/child"): + with pytest.raises(TimeNetMissingParentError, match=r"^parent release test/parent@1\.0\.0 of test/child") as caught: timenet_connectors.load(CHILD.dataset_id) + assert str(caught.value).endswith(". Copy it with fetch_from= or build it with build_parents=True.") if obtain == "fetch_from": dataset = timenet_connectors.load(CHILD.dataset_id, fetch_from=origin) else: @@ -608,7 +580,7 @@ def test_a_fetch_of_a_release_that_is_not_open_fails_before_any_download(tmp_pat output = LocalRegistry(tmp_path / "out") with pytest.raises(TimeNetAccessError) as caught: - prepare_parents(CHILD.dataset_id, output, build=_no_build, fetch_from=origin) + build_release(CHILD.dataset_id, store=output, fetch_from=origin) assert caught.value.dataset == PARENT assert caught.value.requirements == (AccessRequirement(requirement, Access.CREDENTIALED, _GATED["access_url"]),) @@ -632,7 +604,7 @@ def child(command, env): pin_files.extend(Path(argv[at + 1]) for at, arg in enumerate(argv) if arg == "--pinned-by") monkeypatch.setenv("TIMENET_ISOLATION", env["TIMENET_ISOLATION"]) result = runner.invoke(app, argv) - return result.stdout, "", result.exit_code + return "", result.exit_code monkeypatch.delenv("TIMENET_ISOLATION", raising=False) monkeypatch.setattr(env_module, "_run_build", child) @@ -650,122 +622,6 @@ def child(command, env): assert output.exists(GRAND.dataset_id, str(GRAND.version)) -STORED = DatasetRef.parse("test/stored@1.0.0") - - -class _CopyBackend(_CardBackend): - """A second builder backend that rebuilds a release from a saved copy of it, and records each call.""" - - def __init__(self, cards, saved, calls): - super().__init__(cards) - self.saved = saved - self.calls = calls - - def build(self, dataset_id, root, **options): - self.calls.append((dataset_id, options)) - shutil.copytree(self.saved, release_dir(root, STORED)) - return release_dir(root, STORED) - - -@pytest.mark.parametrize("output_kind", ["local", "remote"]) -def test_a_parent_that_another_backend_claims_builds_there_with_its_pins(tmp_path, monkeypatch, output_kind): - output = LocalRegistry(tmp_path / "output") - stored = make_dataset() - stored._metadata = stored.metadata.model_copy(update={"dataset_id": STORED.dataset_id}) - output.put_dataset(stored) - _child_of(output, PARENT, STORED, make_dataset().records[0].id) - pins = output.get_manifest(PARENT.dataset_id).payload_sources[SourceKey(str(STORED))] - saved = tmp_path / "saved" - shutil.move(release_dir(output.root, STORED), saved) - calls, isolated = [], [] - # The in-repo backend comes first but claims none of the test datasets. - backends = [ConnectorBuilder(), _CopyBackend(_cards(child={"parent": PARENT}, parent={}, stored={}), saved, calls)] - install_backends(monkeypatch, *backends) - monkeypatch.delenv("TIMENET_ISOLATION", raising=False) - monkeypatch.setattr(cli_module, "run_isolated", _isolated_builds(isolated)) - if output_kind == "remote": - # A builder backend writes into a local directory, so it cannot build into a remote output. - monkeypatch.setattr(cli_module, "_resolve_target", lambda target: _Hosted(output.root)) - with pytest.raises(TimeFValidationError, match=r"test/stored@1\.0\.0 builds through another builder backend"): - cli_module.build_one(CHILD.dataset_id, output.root, build_parents=True) - assert calls == isolated == [] - return - - cli_module.build_one(CHILD.dataset_id, output.root, build_parents=True) - - # The second backend rebuilds STORED against the pins of PARENT, with no options to plan or fetch - # its parents again. Only the requested build takes the in-repo path. - assert calls == [(STORED.dataset_id, {"pinned_by": {PARENT: pins}})] - assert isolated == [(CHILD.dataset_id, False)] - - -class _OwnBuild(_InRepoCards): - """A subclass of the builder backend of this package that overrides ``build()``, and records each call.""" - - def __init__(self, cards, calls): - super().__init__(cards) - self.calls = calls - - def build(self, dataset_id, root, **options): - self.calls.append((dataset_id, options)) - dataset = make_dataset() - dataset._metadata = self.cards[dataset_id] - LocalRegistry(root).put_dataset(dataset) - return release_dir(root, DatasetRef.parse(f"{dataset_id}@{dataset.metadata.dataset_version}")) - - -@pytest.mark.parametrize("overrides", [True, False], ids=["overrides-build", "inherits-build"]) -def test_a_subclass_of_the_in_repo_backend_builds_a_parent_through_its_own_build(tmp_path, monkeypatch, overrides): - cards = _cards(child={"parent": PARENT}, parent={}) - own, isolated = [], [] - backend = _OwnBuild(cards, own) if overrides else _InRepoCards(cards) - install_backends(monkeypatch, backend) - monkeypatch.delenv("TIMENET_ISOLATION", raising=False) - monkeypatch.setattr(cli_module, "run_isolated", _isolated_builds(isolated)) - output = LocalRegistry(tmp_path / "output") - - cli_module.build_one(CHILD.dataset_id, output.root, build_parents=True) - - assert output.exists(PARENT.dataset_id, str(PARENT.version)) - if overrides: - # The parent builds through the build() of the subclass. Only the requested build takes the - # built-in path. - assert own == [(PARENT.dataset_id, {"pinned_by": None})] - assert isolated == [(CHILD.dataset_id, False)] - else: - # Without an own build(), the parent takes the built-in path, as the requested build does. - assert isolated == [(PARENT.dataset_id, False), (CHILD.dataset_id, False)] - - -class _Proxy: - """A builder backend that is not a ``ConnectorBuilder``, and whose methods come from ``__getattr__``.""" - - def __init__(self, backend): - self._backend = backend - - def __getattr__(self, name): - return getattr(self._backend, name) - - -def test_a_proxy_backend_builds_a_parent_through_the_build_of_its_instance(tmp_path, monkeypatch): - own, isolated = [], [] - # The class of the proxy has no build(). The instance forwards build() to a backend that records it. - backend = _Proxy(_OwnBuild(_cards(child={"parent": PARENT}, parent={}), own)) - assert not hasattr(_Proxy, "build") - install_backends(monkeypatch, backend) - monkeypatch.delenv("TIMENET_ISOLATION", raising=False) - monkeypatch.setattr(cli_module, "run_isolated", _isolated_builds(isolated)) - output = LocalRegistry(tmp_path / "output") - - cli_module.build_one(CHILD.dataset_id, output.root, build_parents=True) - - # The parent builds through the build() that the proxy forwards. Only the requested build takes - # the built-in path. - assert output.exists(PARENT.dataset_id, str(PARENT.version)) - assert own == [(PARENT.dataset_id, {"pinned_by": None})] - assert isolated == [(CHILD.dataset_id, False)] - - @pytest.mark.parametrize("case", ["missing", "damaged", "not-open"]) def test_the_cli_hint_names_its_flags(tmp_path, monkeypatch, case): cards = _install_cards(monkeypatch, child={"parent": PARENT}, parent={}) diff --git a/packages/timenet-connectors/tests/test_discovery.py b/packages/timenet-connectors/tests/test_discovery.py index 759f218b8..aa36b0a0a 100644 --- a/packages/timenet-connectors/tests/test_discovery.py +++ b/packages/timenet-connectors/tests/test_discovery.py @@ -4,7 +4,7 @@ import pytest from timenet_connectors import discovery -from timenet_connectors.discovery import connector_dir, has_connector, requirements_for +from timenet_connectors.discovery import connector_dir, requirements_for def test_connector_dir_points_at_the_connector_package(): @@ -57,33 +57,6 @@ def test_requirements_for_returns_the_declared_file(tmp_path, monkeypatch): assert requirements_for("any/id") == tmp_path / "requirements.txt" -def test_has_connector_accepts_a_known_id(): - assert has_connector("physionet/ecg-qa-cot") - - -def test_has_connector_rejects_an_unknown_id(): - assert not has_connector("nope/nothing") - - -def test_has_connector_rejects_a_bare_org(): - # An org namespace package resolves to a spec with submodule_search_locations, but it holds no - # dataset.yaml, so it is not a connector. - assert not has_connector("physionet") - assert not has_connector("timenet") - assert not has_connector("chengsenwang") - - -def test_has_connector_does_not_import_any_connector_on_a_miss(): - # A miss must not import any connector: their dependencies are absent from this environment - # by design, so importing one raises ModuleNotFoundError instead of answering the question. - leaf = "timenet_connectors.datasets.physionet.ecg_qa_cot" - sys.modules.pop(leaf, None) - - assert not has_connector("nope/nothing") - - assert leaf not in sys.modules - - # A requirement line starts with the distribution name; a version specifier, an extra, a marker, an # inline comment or a direct-reference URL all follow it. _REQUIREMENT_NAME = re.compile(r"^ *([A-Za-z0-9][A-Za-z0-9._-]*)", re.MULTILINE) diff --git a/packages/timenet-connectors/tests/test_end_to_end.py b/packages/timenet-connectors/tests/test_end_to_end.py index 702654e58..64dc2a5d6 100644 --- a/packages/timenet-connectors/tests/test_end_to_end.py +++ b/packages/timenet-connectors/tests/test_end_to_end.py @@ -6,10 +6,10 @@ from timenet.client import TimeNet from timenet.config import settings -from timenet.engine import publish_pipeline, run_pipeline +from timenet.engine import run_pipeline from timenet.errors import TimeFValidationError from timenet.provenance import build_env -from timenet.registry import LocalRegistry, RemoteRegistry +from timenet.registry import RemoteRegistry from timenet.testing import assert_datasets_equal from timenet_connectors import build, load from timenet_connectors.builder.cli import _resolve_target, app as build_app @@ -124,17 +124,6 @@ def test_resolve_target_publishes_to_remote_timenet_registry(clean_env, monkeypa assert isinstance(_resolve_target(None), RemoteRegistry) -def test_publish_pipeline_stores_through_writable_registry(clean_env, converted, tmp_path): - # publish_pipeline hands the converted dataset to registry.put_dataset; a LocalRegistry stands in for a - # remote one, so the publish path is exercised without a network round-trip. - registry = LocalRegistry(tmp_path / "registry") - assert publish_pipeline(HelloWorldConnector(), registry) == "1.0.0" - - restored = TimeNet(tmp_path / "registry", storage_path=tmp_path / "store").load("timenet/hello-world") - [original] = converted - assert_datasets_equal(original, restored) - - def test_run_pipeline_cleans_cache_when_requested(tmp_path, monkeypatch): monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home")) version_dir = run_pipeline(HelloWorldConnector(), tmp_path / "registry", keep_cache=False) diff --git a/packages/timenet/src/timenet/builders.py b/packages/timenet/src/timenet/builders.py index 2f479da66..30d766082 100644 --- a/packages/timenet/src/timenet/builders.py +++ b/packages/timenet/src/timenet/builders.py @@ -9,7 +9,6 @@ from collections.abc import Iterable, Mapping from dataclasses import dataclass from importlib.metadata import entry_points -from pathlib import Path from typing import TYPE_CHECKING, Literal, Protocol from timenet.errors import TimeNetDatasetNotFoundError, TimeNetVersionUnavailableError @@ -20,7 +19,7 @@ if TYPE_CHECKING: from timenet.manifest import PayloadSource - from timenet.registry import BaseRegistry, LocalRegistry + from timenet.registry import LocalRegistry ENTRY_POINT_GROUP = "timenet.builders" @@ -44,7 +43,7 @@ class BuildOptions: class BuilderBackend(Protocol): - """An interface for building dataset releases from connectors.""" + """Define the interface for reading connector cards and building dataset releases.""" def dataset_ids(self) -> Iterable[str]: """List the datasets this backend can build. @@ -59,43 +58,11 @@ def dataset_ids(self) -> Iterable[str]: """ ... - def knows(self, dataset_id: str) -> bool: - """Report whether this backend has a connector for the id. - - Implementations must answer without importing the connector: this runs before any - connector environment exists. - - Args: - dataset_id: The dataset id. - - Returns: - Whether the backend can build it. - """ - ... - - def declared_version(self, dataset_id: str) -> str | None: - """Return the version the connector declares, without importing or building it. - - A caller checks a version pin against this before the build runs, so a pin the connector - cannot satisfy fails fast instead of after a full build. Answer without importing the - connector, like :meth:`knows`. - - Args: - dataset_id: The dataset id. - - Returns: - The declared version string, or ``None`` if the backend has no connector for the id. - - Raises: - TimeNetInvalidCardError: If the card of the connector cannot be read. - """ - ... - def card(self, dataset_id: str) -> DatasetMetadata | None: """Return the card the connector declares, without importing or building it. A build planner reads the card for a parent's exact version, its own parents, and its access - requirement before anything runs. Answer without importing the connector, like :meth:`knows`. + requirement before anything runs. Answer without importing the connector, like :meth:`dataset_ids`. Args: dataset_id: The dataset id. @@ -108,37 +75,7 @@ def card(self, dataset_id: str) -> DatasetMetadata | None: """ ... - def build( # ruff: ignore[too-many-arguments] - self, - dataset_id: str, - root: Path, - *, - force: bool = False, - build_parents: bool = False, - fetch_from: "str | BaseRegistry | None" = None, - pinned_by: "Mapping[DatasetRef, PayloadSource] | None" = None, - ) -> Path: - """Build the dataset into a registry directory. - - Missing parent releases are prepared before the build starts, or the build fails before - any download. Each one is reused, copied from ``fetch_from``, or built when - ``build_parents`` is set. A build with ``pinned_by`` publishes the release only if it - reproduces each pinned file. - - Args: - dataset_id: The dataset id. - root: The output registry directory. - force: Rebuild even if the version is already built. - build_parents: Build missing parent releases from their installed connectors. - fetch_from: The registry, or its URI, to copy missing parent releases from. - pinned_by: The releases that already read files of this release, with their pins. - - Returns: - The committed version directory. - """ - ... - - def build_prepared( + def build( self, release: DatasetRef, store: "LocalRegistry", diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py index a3156a837..6f4234227 100644 --- a/packages/timenet/src/timenet/client.py +++ b/packages/timenet/src/timenet/client.py @@ -16,6 +16,7 @@ from timenet.builders import installed_builder from timenet.config import settings +from timenet.coordinator.build import build_release from timenet.dataset import TimeFDataset from timenet.errors import ( TimeFValidationError, @@ -68,7 +69,7 @@ def _resolve_ref(dataset_id: str, version: str | None) -> tuple[str, str | None] class TimeNet: - """Browse a registry and fetch datasets to local storage.""" + """Browse, download, and read datasets from a registry.""" def __init__( self, registry: str | Path | BaseRegistry | None = None, *, storage_path: str | Path | None = None @@ -234,8 +235,9 @@ def open( TimeNetAccessError: If the dataset is not open access and the registry is not local. Also if the call must copy a release without access declared ``open`` from a registry that is not local. - TimeNetDatasetNotFoundError: If the version is absent and nothing builds it, or the connector - declares a different version than the one asked for. + TimeNetDatasetNotFoundError: If the version is absent and nothing builds it. + TimeNetVersionUnavailableError: If the version is absent and the connector declares another + version. The build does not start then. TimeNetInvalidCardError: If the card of the connector cannot be read. TimeNetMissingParentError: If a release that the version or its build needs is absent and the call cannot obtain it. The message names the keyword that can obtain it. @@ -293,8 +295,9 @@ def load( TimeNetAccessError: If the dataset is not open access and the registry is not local. Also if the call must copy a release without access declared ``open`` from a registry that is not local. - TimeNetDatasetNotFoundError: If the version is absent and nothing builds it, or the connector - declares a different version than the one asked for. + TimeNetDatasetNotFoundError: If the version is absent and nothing builds it. + TimeNetVersionUnavailableError: If the version is absent and the connector declares another + version. The build does not start then. TimeNetInvalidCardError: If the card of the connector cannot be read. TimeNetMissingParentError: If a release that the version or its build needs is absent and the call cannot obtain it. @@ -390,18 +393,19 @@ def _missing_hint(*, auto_build: bool, fetch_from: BaseRegistry | None) -> str | def _prepare( registry: LocalRegistry, dataset_id: str, version: str | None, *, auto_build: bool, fetch_from: BaseRegistry | None ) -> DatasetVersion: - """Prepare a local release and its payload sources. + """Prepare a local dataset version and its payload sources. - Try local copies, then ``fetch_from``, then a build if ``auto_build`` is true. - A built payload source must match the files that the release pins. + This function reuses local releases, tries ``fetch_from``, then builds missing releases if + ``auto_build`` is true. Each rebuilt payload source must match the version and file pins the + dataset requires. Returns: The release with each payload source attached. Raises: - TimeNetMissingParentError: If a payload source is missing and cannot be copied or built. - TimeNetInvalidCardError: If a connector's dataset card cannot be read. - """ # ruff: ignore[docstring-extraneous-exception] - raised by _obtain and installed_builder + TimeNetMissingParentError: If copying and building cannot provide a missing payload source. + TimeNetInvalidCardError: If reading a connector's card fails. + """ # ruff: ignore[docstring-extraneous-exception] - raised by _obtain, claim and build_release try: manifest = registry.get_manifest(dataset_id, version) except TimeNetDatasetNotFoundError as miss: @@ -420,14 +424,11 @@ def _prepare( found = installed_builder(source.dataset_id) if auto_build and source not in built else None if found is None or found.release != source: raise - builder = found.backend pins = next(pinned for pinned in manifest.payload_sources.values() if pinned.release == source) - builder.build( - source.dataset_id, - registry.root, - build_parents=True, - fetch_from=fetch_from, - pinned_by={reference: pins}, + # TODO(adrian): Remove temporary glue which maps fetch_from and auto_build onto the + # coordinator until the client takes a list of sources. + build_release( + str(source), store=registry, fetch_from=fetch_from, build_parents=True, pinned_by={reference: pins} ) built.add(source) @@ -441,37 +442,36 @@ def _obtain( # ruff: ignore[too-many-arguments] auto_build: bool, fetch_from: BaseRegistry | None, ) -> Manifest: - """Find a version that the local registry lacks in ``fetch_from``, or build it. + """Get a missing version's manifest from another registry or a local build. + + Try ``fetch_from`` first, then build if ``auto_build`` is true. Returns: - The manifest from ``fetch_from`` or the local build. + The manifest from ``fetch_from`` without copying its files, or the local manifest after a build. Raises: - TimeNetDatasetNotFoundError: If neither source provides the release or the connector declares another version. - TimeNetInvalidCardError: If the connector's card cannot be read. - """ # ruff: ignore[docstring-extraneous-exception] - raised by declared_version + TimeNetDatasetNotFoundError: If neither ``fetch_from`` nor a connector provides the version. + TimeNetVersionUnavailableError: If the connector declares a different version. + TimeNetInvalidCardError: If reading the connector's card fails. + """ # ruff: ignore[docstring-extraneous-exception] - raised by build_release if fetch_from is not None: try: return fetch_from.get_manifest(dataset_id, version) except TimeNetDatasetNotFoundError: pass - found = installed_builder(dataset_id) if auto_build else None - if found is None: + if not auto_build or installed_builder(dataset_id) is None: hint = _missing_hint(auto_build=auto_build, fetch_from=fetch_from) if hint is None: raise miss raise TimeNetDatasetNotFoundError(f"{miss}. {hint}") from miss - builder = found.backend - # A connector produces one declared version. Reject a pin it cannot satisfy before the - # build runs, so a wrong pin fails fast instead of after a full build. Sentinels are - # not pins, so a fresh build satisfies them. - if not is_latest(version): - declared = builder.declared_version(dataset_id) - if declared is not None and version != declared: - raise TimeNetDatasetNotFoundError( - f"the connector for {dataset_id!r} builds version {declared}, not the requested {version}" - ) from miss - # The build prepares the declared parents of the version in the same order: reuse, copy, then - # build. - builder.build(dataset_id, registry.root, build_parents=True, fetch_from=fetch_from) + # TODO(adrian): Remove temporary glue which maps fetch_from and auto_build onto the coordinator + # until the client takes a list of sources. The build prepares the declared parents of the + # version in the same order: reuse, copy, then build. Sentinels such as "latest" are not pins, + # so a fresh build satisfies them. + build_release( + dataset_id if is_latest(version) else f"{dataset_id}@{version}", + store=registry, + fetch_from=fetch_from, + build_parents=True, + ) return registry.get_manifest(dataset_id, version) diff --git a/packages/timenet/src/timenet/coordinator/build.py b/packages/timenet/src/timenet/coordinator/build.py index 1a913442f..a58b69b67 100644 --- a/packages/timenet/src/timenet/coordinator/build.py +++ b/packages/timenet/src/timenet/coordinator/build.py @@ -139,7 +139,7 @@ def _build(step: Step, store: LocalRegistry, *, force: bool, options: BuildOptio builder = declared(step.release.dataset_id, step.release.version).backend if options.progress is not None: options.progress(StepStarted(step.release, "build")) - builder.build_prepared(step.release, store, force=force, pinned_by=step.pinned_by, options=options) + builder.build(step.release, store, force=force, pinned_by=step.pinned_by, options=options) # After a successful build, the local registry must hold the exact release with its payload sources. prepare_local_copy(store.root, step.release) if options.progress is not None: diff --git a/packages/timenet/src/timenet/coordinator/plan.py b/packages/timenet/src/timenet/coordinator/plan.py index 4600b620f..cf4d0fac3 100644 --- a/packages/timenet/src/timenet/coordinator/plan.py +++ b/packages/timenet/src/timenet/coordinator/plan.py @@ -1,9 +1,9 @@ -"""Plan and prepare the parent releases needed to build a dataset. +"""Plan how to obtain the parent releases that a build needs, before any connector runs. -Each :class:`Step` copies or builds one release after the releases it needs. +Each :class:`Step` copies or builds one release after its dependencies are ready. """ -from collections.abc import Callable, Mapping +from collections.abc import Mapping from dataclasses import dataclass, field from graphlib import CycleError, TopologicalSorter from typing import Literal @@ -20,10 +20,9 @@ ) from timenet.manifest import Manifest, PayloadSource from timenet.registry.base import BaseRegistry -from timenet.registry.dependencies import check_file_sizes, prepare_local_copy +from timenet.registry.dependencies import check_file_sizes from timenet.registry.local import LocalRegistry from timenet.registry.version import DatasetVersion -from timenet.registry.writable import WritableRegistry from timenet.types import DatasetMetadata from timenet.types.access import require_open_access from timenet.types.references import DatasetRef @@ -73,7 +72,7 @@ class _Release: class _Graph: """The releases a build needs and the dependencies between them.""" - output: WritableRegistry + output: LocalRegistry fetch_from: BaseRegistry | None build_parents: bool open_only: bool @@ -198,7 +197,7 @@ def steps(self, requested: DatasetRef) -> list[Step]: def plan_parents( # ruff: ignore[too-many-arguments] dataset_id: str, - output: WritableRegistry, + output: LocalRegistry, *, fetch_from: BaseRegistry | None = None, build_parents: bool = False, @@ -208,12 +207,11 @@ def plan_parents( # ruff: ignore[too-many-arguments] """Plan the copy and build steps for a dataset's parents. Copy missing parents and their payload sources from ``fetch_from``, or build them if enabled. - Remote registries and ``open_access_only`` require open access for releases and their lineage. Args: dataset_id: The dataset to build. - output: The registry that the build writes to. - fetch_from: The registry to copy missing parents from for a local output, or ``None``. + output: The local registry that the build writes to. + fetch_from: The registry to copy missing parents from, or ``None``. build_parents: Build missing parents with their installed connectors. force_build: Plan parents for a rebuild even if the requested release already exists. open_access_only: Require ``access: open`` for each release, including the requested release. @@ -224,23 +222,16 @@ def plan_parents( # ruff: ignore[too-many-arguments] Raises: LookupError: If no installed connector declares a card for ``dataset_id``. TimeNetInvalidCardError: If an installed connector's card cannot be read. - TimeFValidationError: If parents form a cycle, versions conflict, or a copy targets a remote output. + TimeFValidationError: If parents form a cycle or versions conflict. TimeNetMissingParentError: If a parent or payload source is missing and ``build_parents`` is false. TimeNetDatasetNotFoundError: If a parent is missing and no installed connector builds it. TimeNetFileError: If a local parent is damaged and no replacement is available. TimeNetAccessError: If a release or its lineage lacks required open access. """ # ruff: ignore[docstring-extraneous-exception] - raised by the helpers of the walk - if fetch_from is not None and not isinstance(output, LocalRegistry): - # A registry that is not local accepts a child only if it already holds the parents. A copy - # step writes a parent only to a local directory, never to such a registry. - raise TimeFValidationError( - "copying parents from a registry needs a local output; publish the parents to the registry first" - ) card = find_card(dataset_id) if card is None: raise LookupError(f"no installed connector declares a card for {dataset_id!r}") - open_only = open_access_only or output.requires_open_access - if open_only: + if open_access_only: require_open_access(card) # For a dataset without parents, the plan does not read the output, so the build makes no extra request. if not card.parents: @@ -249,11 +240,11 @@ def plan_parents( # ruff: ignore[too-many-arguments] # If the output already holds the requested release, a reader needs only its payload sources to # open it. Its declared parents matter only to a rebuild. held = None if force_build else _manifest(output, root) - if held is not None and open_only: + if held is not None and open_access_only: # The installed card has no lineage, so the check above does not see an ancestor that is not open. require_open_access(held.metadata, held.lineage) searched = str(output) if fetch_from is None else f"{output} or {fetch_from}" - graph = _Graph(output, fetch_from, build_parents=build_parents, open_only=open_only, searched=searched) + graph = _Graph(output, fetch_from, build_parents=build_parents, open_only=open_access_only, searched=searched) # The graph holds the requested release before the walk starts. So a release that depends on it # closes a cycle, and the walk does not look it up again. if held is None: @@ -270,48 +261,6 @@ def plan_parents( # ruff: ignore[too-many-arguments] return graph.steps(root) -def prepare_parents( # ruff: ignore[too-many-arguments] - dataset_id: str, - output: WritableRegistry, - *, - build: Callable[[Step], object], - fetch_from: BaseRegistry | None = None, - build_parents: bool = False, - force_build: bool = False, -) -> None: - """Copy or build a dataset's required parent releases. - - The ``build`` callback must publish each release only if its files match the step's - ``pinned_by`` records. - - Args: - dataset_id: The dataset to build. - output: The destination registry. - build: The function that builds one parent into ``output``. - fetch_from: The registry to copy missing parents from, or ``None`` to copy nothing. - build_parents: Build missing parents with their installed connectors. - force_build: Prepare a rebuild's parents even if ``output`` holds the requested release. - - Raises: - TimeNetDatasetNotFoundError: If ``build`` does not publish the release of its step. - """ - steps = plan_parents( - dataset_id, output, fetch_from=fetch_from, build_parents=build_parents, force_build=force_build - ) - for step in steps: - if step.action == "copy": - prepare_local_copy(output.local_root, step.release, fetch_from=fetch_from) - else: - build(step) - # After a successful build, the output must hold the exact release with its payload sources. - if isinstance(output, LocalRegistry): - prepare_local_copy(output.root, step.release) - else: - if _manifest(output, step.release) is None: - raise TimeNetDatasetNotFoundError(f"builder did not publish parent release {step.release}") - output.open_version(step.release.dataset_id, str(step.release.version)) - - def _manifest(registry: BaseRegistry, reference: DatasetRef) -> Manifest | None: """Read the manifest of one release. @@ -333,7 +282,7 @@ def _manifest(registry: BaseRegistry, reference: DatasetRef) -> Manifest | None: def _locate( - output: WritableRegistry, fetch_from: BaseRegistry | None, reference: DatasetRef, *, open_only: bool + output: LocalRegistry, fetch_from: BaseRegistry | None, reference: DatasetRef, *, open_only: bool ) -> tuple[Manifest | None, Literal["copy"] | None]: # The damage is raised from a variable, the access error by require_open_access. # ruff: ignore[docstring-extraneous-exception] @@ -351,7 +300,7 @@ def _locate( """ manifest = _manifest(output, reference) damage = None - if manifest is not None and isinstance(output, LocalRegistry): + if manifest is not None: directory = output.root / reference.dataset_id / str(reference.version) try: check_file_sizes( diff --git a/packages/timenet/src/timenet/engine/__init__.py b/packages/timenet/src/timenet/engine/__init__.py index 76575485c..dbf8857ad 100644 --- a/packages/timenet/src/timenet/engine/__init__.py +++ b/packages/timenet/src/timenet/engine/__init__.py @@ -1,6 +1,6 @@ """The build engine: drive a connector through ``download -> compose -> derive_schema -> store``.""" -from timenet.engine.engine import open_parents, publish_pipeline, run_pipeline, store_dataset +from timenet.engine.engine import open_parents, run_pipeline, store_dataset -__all__ = ["open_parents", "publish_pipeline", "run_pipeline", "store_dataset"] +__all__ = ["open_parents", "run_pipeline", "store_dataset"] diff --git a/packages/timenet/src/timenet/engine/engine.py b/packages/timenet/src/timenet/engine/engine.py index f34c8d339..22a0e4e99 100644 --- a/packages/timenet/src/timenet/engine/engine.py +++ b/packages/timenet/src/timenet/engine/engine.py @@ -13,12 +13,9 @@ from timenet.format.constants import MANIFEST_FILE from timenet.manifest import PayloadSource from timenet.reader import TimeFReader -from timenet.registry.base import BaseRegistry from timenet.registry.dependencies import prepare_local_copy from timenet.registry.local import LocalRegistry -from timenet.registry.writable import WritableRegistry from timenet.types import DatasetMetadata -from timenet.types.access import require_open_access from timenet.types.reference_keys import ParentAlias from timenet.types.references import DatasetRef from timenet.values_backends import SUPPORTED_VALUES_BACKENDS, ValuesBackend @@ -103,73 +100,15 @@ def run_pipeline( # ruff: ignore[too-many-arguments] return version_dir -def publish_pipeline( # ruff: ignore[too-many-arguments] - connector: BaseConnector, - registry: WritableRegistry, - *, - cache_dir: Path | None = None, - keep_cache: bool = False, - progress_cb: Callable[[WriteProgressEvent], None] | None = None, - force: bool = False, - pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, -) -> str: - """Build and publish a dataset to a writable registry. - - Reuse an existing version unless ``force`` is set. - The dataset card and lineage must declare open access if the registry requires it. - - Args: - connector: The connector to build. - registry: The writable registry to publish to. - cache_dir: The download directory, or ``None`` for ``/``. - keep_cache: Keep the default cache after publishing. A supplied ``cache_dir`` is always kept. - progress_cb: Callback for writer progress events, or ``None``. - force: Publish again even if the version already exists. - pinned_by: Releases that read this release's files, mapped to file records the build must match. - - Returns: - The existing or newly published version string. - - Raises: - TimeFValidationError: If a connector declares parents without overriding ``compose``, or changes its card. - TimeNetAccessError: If the card, dataset, or lineage lacks open access and the registry requires it. - """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_composes, _compose, and require_open_access - values_backend = _check_values_backend(connector) - metadata = connector.metadata() - dataset_id = metadata.dataset_id - version = str(metadata.dataset_version) - if not force and registry.exists(dataset_id, version): - return version - if registry.requires_open_access: - require_open_access(metadata) - _check_composes(connector, metadata) - - cache = cache_dir if cache_dir is not None else settings().cache_dir / dataset_id - cache.mkdir(parents=True, exist_ok=True) - - with open_parents(metadata, registry) as context: - dataset = _compose(connector, metadata, cache, context) - if registry.requires_open_access: - require_open_access(dataset.metadata, dataset.lineage.values()) - dataset.derive_schema() - registry.put_dataset( - dataset, force=force, values_backend=values_backend, progress_cb=progress_cb, pinned_by=pinned_by - ) - if not keep_cache and cache_dir is None and cache.is_dir(): - shutil.rmtree(cache) - return version - - -def open_parents(metadata: DatasetMetadata, registry: BaseRegistry) -> BuildContext: - """Open each declared parent of a card from the registry a build writes to. +def open_parents(metadata: DatasetMetadata, registry: LocalRegistry) -> BuildContext: + """Open readers for the dataset's declared parents in a local registry. - A local registry must hold every parent complete, with its payload sources. A remote registry - resolves the payload source manifests upfront, and the readers read catalogs and payloads only - when the connector needs them. The engine never fetches or builds a parent here. + The registry must hold every parent complete, with its payload sources. The engine never fetches + or builds a parent here. Args: metadata: The child dataset card with exact parent references. - registry: The registry the child is written to. + registry: The local registry containing the parents. Returns: The opened readers, keyed by parent alias. Close the context after the write. @@ -182,12 +121,7 @@ def open_parents(metadata: DatasetMetadata, registry: BaseRegistry) -> BuildCont # the stack lets the readers go, and the context closes them. with ExitStack() as stack: for alias, ref in metadata.parents.items(): - version = ( - prepare_local_copy(registry.root, ref) - if isinstance(registry, LocalRegistry) - else registry.open_version(ref.dataset_id, str(ref.version)) - ) - parents[alias] = stack.enter_context(TimeFReader(version)) + parents[alias] = stack.enter_context(TimeFReader(prepare_local_copy(registry.root, ref))) stack.pop_all() return BuildContext(references=dict(metadata.parents), parents=parents) diff --git a/packages/timenet/src/timenet/registry/lazy_read.py b/packages/timenet/src/timenet/registry/lazy_read.py index 0b6c944ac..dca91c38b 100644 --- a/packages/timenet/src/timenet/registry/lazy_read.py +++ b/packages/timenet/src/timenet/registry/lazy_read.py @@ -8,6 +8,7 @@ from typing import ClassVar from timenet.dataset import TimeFDataset +from timenet.errors import TimeFValidationError from timenet.manifest import PayloadSource from timenet.registry._paths import release_dir from timenet.registry.dependencies import open_cached_or_remote @@ -38,24 +39,31 @@ def put_dataset( ) -> DatasetRef: """Write a dataset locally and publish it to this registry. - Return the existing release's reference unless ``force`` is set. + If the version already exists, return its reference. A published version never changes. Args: dataset: The dataset to write. - force: Publish even if the version already exists. + force: Must be false, because this registry never writes a published version again. values_backend: Storage format for values (``"parquet"`` or ``"zarr"``). progress_cb: Optional callback for write progress. pinned_by: Releases that already read this release's files, with the pins that the new files must match. Returns: A reference to the stored release. + + Raises: + TimeFValidationError: If ``force`` is set. """ + if force: + raise TimeFValidationError( + f"force=True needs a local registry: {self} never writes a published version again" + ) if dataset.schema is None: dataset.derive_schema() dataset_id = dataset.metadata.dataset_id version = str(dataset.metadata.dataset_version) reference = DatasetRef.parse(f"{dataset_id}@{version}") - if not force and self.exists(dataset_id, version): + if self.exists(dataset_id, version): return reference staging = Path(tempfile.mkdtemp(prefix=self._staging_prefix)) try: diff --git a/packages/timenet/tests/coordinator/test_build.py b/packages/timenet/tests/coordinator/test_build.py index 69b4fde41..be5ca53cc 100644 --- a/packages/timenet/tests/coordinator/test_build.py +++ b/packages/timenet/tests/coordinator/test_build.py @@ -60,7 +60,7 @@ def dataset_ids(self): def card(self, dataset_id): return self.cards.get(dataset_id) - def build_prepared(self, release, store, *, force, pinned_by, options): + def build(self, release, store, *, force, pinned_by, options): self.calls.append((release, force, dict(pinned_by))) store.put_dataset(_demo(self.cards[release.dataset_id]), force=force, pinned_by=pinned_by or None) diff --git a/packages/timenet/tests/registry/test_conformance.py b/packages/timenet/tests/registry/test_conformance.py index 1fe6a7873..f365f56ca 100644 --- a/packages/timenet/tests/registry/test_conformance.py +++ b/packages/timenet/tests/registry/test_conformance.py @@ -199,10 +199,6 @@ def test_put_dataset_is_idempotent(backend): assert backend.registry.put_dataset(make_dataset()) == DatasetRef.parse("timenet/hello-world@1.0.0") -def test_put_dataset_force_republishes(backend): - assert backend.registry.put_dataset(make_dataset(), force=True) == DatasetRef.parse("timenet/hello-world@1.0.0") - - def _local_release(tmp_path): """Build a complete local release of timenet/hello-world 2.0.0; return its directory and manifest.""" source = LocalRegistry(tmp_path / "src") diff --git a/packages/timenet/tests/registry/test_remote.py b/packages/timenet/tests/registry/test_remote.py index 69e96c072..8ae6242c4 100644 --- a/packages/timenet/tests/registry/test_remote.py +++ b/packages/timenet/tests/registry/test_remote.py @@ -13,6 +13,7 @@ from timenet.dataset import Record, RegularAxis, Signal, Source, TimeFDataset from timenet.errors import ( TimeFFormatError, + TimeFValidationError, TimeNetAccessError, TimeNetDatasetNotFoundError, TimeNetFileError, @@ -385,6 +386,15 @@ def test_put_dataset_publishes_uploads_and_finalizes(tmp_path): assert set(state["store"]) == set(state["published"]) +def test_put_dataset_refuses_force_and_uploads_nothing(tmp_path): + transport, state = build_publish_fake(token="tok_rw") + registry = RemoteRegistry("http://api.local", token="tok_rw", transport=transport, cache_dir=tmp_path) + with pytest.raises(TimeFValidationError, match="never writes a published version again"): + registry.put_dataset(make_dataset(), force=True) + assert state["manifest"] is None + assert not state["store"] + + def test_put_dataset_refuses_a_payload_source_the_service_does_not_hold_before_publishing(tmp_path): local = LocalRegistry(tmp_path / "local") local.put_dataset(make_dataset()) @@ -413,6 +423,9 @@ def test_put_dataset_rejects_a_publish_list_that_drops_a_manifest_file(tmp_path) def handler(request): # The service omits every declared file from its upload list. put_dataset() must reject that before # uploading, rather than publish an incomplete version. + if request.method == "GET": + # The service does not hold the version yet. + return httpx.Response(404) if request.url.path.endswith("/publish"): return httpx.Response(200, json={"files": []}) return httpx.Response(200, json={}) @@ -421,11 +434,14 @@ def handler(request): "http://api.local", token="tok_rw", transport=httpx.MockTransport(handler), cache_dir=tmp_path ) with pytest.raises(TimeNetRegistryError, match="disagrees with the manifest"): - registry.put_dataset(make_dataset(), force=True) + registry.put_dataset(make_dataset()) def test_put_dataset_rejects_unsafe_publish_paths(tmp_path): def handler(request): + if request.method == "GET": + # The service does not hold the version yet. + return httpx.Response(404) if request.url.path.endswith("/publish"): return httpx.Response(200, json={"files": ["../escape"]}) return httpx.Response(200, json={}) @@ -434,11 +450,14 @@ def handler(request): "http://api.local", token="tok_rw", transport=httpx.MockTransport(handler), cache_dir=tmp_path ) with pytest.raises(ValidationError, match="stay within the dataset root"): - registry.put_dataset(make_dataset(), force=True) + registry.put_dataset(make_dataset()) def test_put_dataset_rejects_malformed_upload_grant(tmp_path): def handler(request): + if request.method == "GET": + # The service does not hold the version yet. + return httpx.Response(404) if request.url.path.endswith("/publish"): manifest = json.loads(request.content) files = [part["path"] for group in manifest["files"].values() for part in group] @@ -451,7 +470,7 @@ def handler(request): "http://api.local", token="tok_rw", transport=httpx.MockTransport(handler), cache_dir=tmp_path ) with pytest.raises(ValidationError, match="headers"): - registry.put_dataset(make_dataset(), force=True) + registry.put_dataset(make_dataset()) def test_latest_detail_rejects_non_string_version(tmp_path): diff --git a/packages/timenet/tests/test_builders.py b/packages/timenet/tests/test_builders.py index ca0edeebc..0809f5fea 100644 --- a/packages/timenet/tests/test_builders.py +++ b/packages/timenet/tests/test_builders.py @@ -17,7 +17,7 @@ def dataset_ids(self): def card(self, dataset_id): return self._card - def build_prepared(self, release, store, *, force, pinned_by, options): + def build(self, release, store, *, force, pinned_by, options): raise AssertionError(f"the lookup does not build {release}") diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py index 785c6e160..2307858d7 100644 --- a/packages/timenet/tests/test_client.py +++ b/packages/timenet/tests/test_client.py @@ -6,8 +6,7 @@ from pydantic import ValidationError import pytest -from timenet import client as client_module -from timenet.builders import InstalledBuilder +from timenet import builders as builders_module, client as client_module from timenet.client import TimeNet from timenet.connectors import BaseConnector from timenet.dataset import Record, RegularAxis, Signal, Source, TimeFDataset @@ -19,6 +18,7 @@ TimeNetDatasetNotFoundError, TimeNetFileError, TimeNetMissingParentError, + TimeNetVersionUnavailableError, ) from timenet.format.stored_values import StoredText from timenet.manifest import Manifest @@ -71,28 +71,32 @@ def __init__(self, calls): def dataset_ids(self): return ("test/child", "timenet/hello-world") - def knows(self, dataset_id): - return True - - def declared_version(self, dataset_id): - return self.VERSION - def card(self, dataset_id): version = DatasetRef.parse(f"{dataset_id}@{self.VERSION}").version return make_dataset().metadata.model_copy(update={"dataset_id": dataset_id, "dataset_version": version}) - # ruff: ignore[too-many-arguments] - def build(self, dataset_id, root, *, force=False, build_parents=False, fetch_from=None, pinned_by=None): - self.calls.append((dataset_id, force, build_parents, fetch_from, pinned_by)) + def build(self, release, store, *, force, pinned_by, options): + self.calls.append((release, force, dict(pinned_by))) dataset = make_dataset() - dataset._metadata = self.card(dataset_id) - LocalRegistry(root).put_dataset(dataset, force=force, values_backend=self.BACKEND, pinned_by=pinned_by) - return root / dataset_id / self.VERSION + dataset._metadata = self.card(release.dataset_id) + store.put_dataset(dataset, force=force, values_backend=self.BACKEND, pinned_by=pinned_by or None) + + +def _record_build_requests(monkeypatch): + """Record the dataset, ``fetch_from`` and ``build_parents`` of each build that the client asks for. + + Returns: + The list that receives one ``(dataset, fetch_from, build_parents)`` entry for each request. + """ + requests = [] + build_release = client_module.build_release + def record(dataset, **options): + requests.append((dataset, options.get("fetch_from"), options.get("build_parents"))) + return build_release(dataset, **options) -def _installed_builders(builder, card=None): - """Return a stand-in for ``installed_builder`` that answers every dataset with ``builder``, and with ``card`` or its card.""" - return lambda dataset_id: InstalledBuilder(builder, card or builder.card(dataset_id)) + monkeypatch.setattr(client_module, "build_release", record) + return requests def test_a_local_child_copies_its_missing_payload_source_from_fetch_from(tmp_path): @@ -121,17 +125,19 @@ def test_a_missing_version_is_copied_from_fetch_from_before_it_is_built(tmp_path if in_fetch_from: _child_of_the_demo(origin.root) calls = [] - monkeypatch.setattr(client_module, "installed_builder", _installed_builders(_Builder(calls))) + install_backends(monkeypatch, _Builder(calls)) + requests = _record_build_requests(monkeypatch) with TimeNet(tmp_path / "local").open("test/child", fetch_from=origin) as reader: assert reader.read().records if in_fetch_from: - assert calls == [] + assert calls == requests == [] assert (tmp_path / "local/test/child/1.0.0/manifest.json").is_file() return # The build gets the same registry to copy its own missing parents from. - assert calls == [("test/child", False, True, origin, None)] + assert requests == [("test/child", origin, True)] + assert calls == [(DatasetRef.parse("test/child@1.0.0"), False, {})] class _Resampled(BaseConnector[str]): @@ -197,6 +203,7 @@ def test_a_missing_payload_source_builds_with_its_own_connector_against_the_pins monkeypatch.setattr(_Builder, "VERSION", "2.0.0" if case == "other-version" else "1.0.0") monkeypatch.setattr(_Builder, "BACKEND", "zarr" if case == "differs" else "parquet") install_backends(monkeypatch, _Builder(calls)) + requests = _record_build_requests(monkeypatch) with pytest.raises(TimeNetMissingParentError, match=r"auto_build=True\.$"): TimeNet(registry).open("test/child", auto_build=False) @@ -205,7 +212,7 @@ def test_a_missing_payload_source_builds_with_its_own_connector_against_the_pins # No installed connector declares the exact version of the source that the child pins. with pytest.raises(TimeNetMissingParentError, match=r"\. Copy it with fetch_from=\.$"): TimeNet(registry).open("test/child") - assert calls == [] + assert calls == requests == [] return if case == "differs": pytest.importorskip("zarr") @@ -222,7 +229,8 @@ def test_a_missing_payload_source_builds_with_its_own_connector_against_the_pins assert signal.to_arrow().equals(make_dataset().records[0].signals[0].to_arrow()) # The source builds with its own connector, which prepares its parents in the same order, against # the pins of the child that reads it. - assert calls == [(reference.dataset_id, False, True, fetch_from, {DatasetRef.parse("test/child@1.0.0"): pins})] + assert requests == [(str(reference), fetch_from, True)] + assert calls == [(reference, False, {DatasetRef.parse("test/child@1.0.0"): pins})] @pytest.mark.parametrize("damaged", ["requested", "payload-source"]) @@ -236,7 +244,6 @@ def test_a_damaged_release_is_repaired_from_fetch_from_and_never_built_over(tmp_ control.write_bytes(b"cut short") calls = [] install_backends(monkeypatch, _Builder(calls)) - monkeypatch.setattr(client_module, "installed_builder", _installed_builders(_Builder(calls))) with pytest.raises(TimeNetFileError) as caught: TimeNet(registry).open("test/child") @@ -265,7 +272,7 @@ def test_a_corrupted_copy_from_fetch_from_names_the_transfer_without_the_repair_ def test_a_remote_client_takes_no_fetch_from_and_never_builds(tmp_path, monkeypatch): calls = [] - monkeypatch.setattr(client_module, "installed_builder", _installed_builders(_Builder(calls))) + install_backends(monkeypatch, _Builder(calls)) remote = RemoteRegistry( "http://api.local", transport=httpx.MockTransport(lambda request: httpx.Response(404)), cache_dir=tmp_path ) @@ -550,49 +557,40 @@ def test_version_given_twice_raises(versioned_registry, tmp_path): def test_load_names_the_buildable_version_when_a_pin_cannot_be_built(registry_root, tmp_path, monkeypatch): # The builder declares only one version, so a pin it cannot satisfy must say which version it # builds, and must fail before the expensive build rather than after it. - class _Builder: - built = False + calls = [] + install_backends(monkeypatch, _Builder(calls)) + client = TimeNet(registry_root, storage_path=tmp_path / "store") - def knows(self, dataset_id): - return True + with pytest.raises(TimeNetVersionUnavailableError, match=r"builds version 1\.0\.0, not the requested 9\.9\.9"): + client.load("timenet/hello-world@9.9.9") + assert calls == [] # the pin was rejected before the build ran - def declared_version(self, dataset_id): - return "1.0.0" - def build(self, dataset_id, root, *, force=False, build_parents=False, fetch_from=None): - self.built = True - return Path(root) / dataset_id / "1.0.0" +@pytest.mark.parametrize("fetch_from", [None, "empty"]) +def test_a_missing_version_that_no_connector_builds_names_the_keyword_that_can_get_it( + tmp_path, monkeypatch, fetch_from +): + monkeypatch.setattr(builders_module, "entry_points", lambda group: []) + client = TimeNet(tmp_path / "reg", storage_path=tmp_path / "store") + with pytest.raises(TimeNetDatasetNotFoundError) as miss: + LocalRegistry(tmp_path / "reg").get_manifest("x/y") + source = None if fetch_from is None else LocalRegistry(tmp_path / fetch_from) - builder = _Builder() - monkeypatch.setattr(client_module, "installed_builder", _installed_builders(builder, make_dataset().metadata)) - client = TimeNet(registry_root, storage_path=tmp_path / "store") + with pytest.raises(TimeNetDatasetNotFoundError) as caught: + client.open("x/y", fetch_from=source) - with pytest.raises(TimeNetDatasetNotFoundError, match=r"builds version 1\.0\.0, not the requested 9\.9\.9"): - client.load("timenet/hello-world@9.9.9") - assert not builder.built # the pin was rejected before the build ran + # The hint names only the keyword that the call did not use. + expected = f"{miss.value}. Copy it with fetch_from=." if source is None else str(miss.value) + assert str(caught.value) == expected @pytest.mark.parametrize("pin", [None, "latest", ""]) def test_load_accepts_the_latest_sentinels_after_a_build(tmp_path, monkeypatch, pin): # Every registry reads "latest" and "" as the latest version, so they are not pins the build # can miss: the version the builder just committed satisfies them. - class _Builder: - def knows(self, dataset_id): - return True - - def declared_version(self, dataset_id): - return "1.0.0" - - def build(self, dataset_id, root, *, force=False, build_parents=False, fetch_from=None): - dataset = make_dataset() - dataset.derive_schema() - with TimeFWriter(Path(root), dataset) as writer: - writer.write() - return Path(root) / dataset_id / "1.0.0" - empty_root = tmp_path / "reg" empty_root.mkdir() - monkeypatch.setattr(client_module, "installed_builder", _installed_builders(_Builder(), make_dataset().metadata)) + install_backends(monkeypatch, _Builder([])) client = TimeNet(empty_root, storage_path=tmp_path / "store") assert str(client.load("timenet/hello-world", pin).metadata.dataset_version) == "1.0.0" @@ -603,30 +601,18 @@ def test_load_does_not_build_when_auto_build_is_false(registry_root, tmp_path, m # A caller can opt out of the build-on-miss, so a notebook load fails fast instead of starting a # multi-GB download and a build. The builder declares the requested version, so only # auto_build=False keeps it from running. - class _Builder: - built = False - - def knows(self, dataset_id): - return True - - def declared_version(self, dataset_id): - return "9.9.9" - - def build(self, dataset_id, root, *, force=False, build_parents=False, fetch_from=None): - self.built = True - return Path(root) / dataset_id / "1.0.0" - if entry_point == "open_torch": pytest.importorskip("torch") - builder = _Builder() - monkeypatch.setattr(client_module, "installed_builder", _installed_builders(builder, make_dataset().metadata)) + calls = [] + monkeypatch.setattr(_Builder, "VERSION", "9.9.9") + install_backends(monkeypatch, _Builder(calls)) client = TimeNet(registry_root, storage_path=tmp_path / "store") # The hint names the keywords that could obtain the release, in a sentence of its own. hint = r"\. Copy it with fetch_from= or build it from its connector with auto_build=True\.$" with pytest.raises(TimeNetDatasetNotFoundError, match=hint): getattr(client, entry_point)("timenet/hello-world@9.9.9", auto_build=False) - assert not builder.built # auto_build=False skips the connector entirely + assert calls == [] # auto_build=False skips the connector entirely def _card(dataset_id, access, access_url=None): diff --git a/packages/timenet/tests/test_engine.py b/packages/timenet/tests/test_engine.py index 9efecaa79..0dd8fdee5 100644 --- a/packages/timenet/tests/test_engine.py +++ b/packages/timenet/tests/test_engine.py @@ -2,21 +2,19 @@ import tempfile from typing import Any, ClassVar, cast -import httpx import pytest from timenet.config import settings from timenet.connectors import BaseConnector, BuildContext from timenet.dataset import TimeFDataset -from timenet.engine import open_parents, publish_pipeline, run_pipeline, store_dataset -from timenet.errors import TimeFValidationError, TimeNetAccessError, TimeNetDatasetNotFoundError +from timenet.engine import open_parents, run_pipeline, store_dataset +from timenet.errors import TimeFValidationError, TimeNetDatasetNotFoundError from timenet.manifest import Manifest from timenet.reader import TimeFReader -from timenet.registry import LocalRegistry, RemoteRegistry -from timenet.registry.writable import WritableRegistry +from timenet.registry import LocalRegistry from timenet.testing import make_dataset from timenet.types import Access, Version -from timenet.types.access import AccessRequirement, effective_access +from timenet.types.access import effective_access from timenet.types.reference_keys import ParentAlias from timenet.types.references import DatasetRef from timenet.values_backends import ValuesBackend @@ -104,29 +102,6 @@ class _ZarrConnector(_DemoConnector): assert manifest.values_backend == "zarr" -def test_publish_pipeline_passes_the_requested_values_backend_to_the_registry(tmp_path): - class _Registry: - requires_open_access = True - - def __init__(self) -> None: - self.values_backend: str | None = None - - def exists(self, dataset_id: str, version: str) -> bool: - return False - - def put_dataset(self, dataset, *, force: bool, values_backend: str, progress_cb, pinned_by) -> DatasetRef: - self.values_backend = values_backend - return DatasetRef.parse(f"{dataset.metadata.dataset_id}@{dataset.metadata.dataset_version}") - - class _ZarrConnector(_DemoConnector): - values_backend = ValuesBackend.ZARR - - registry = _Registry() - publish_pipeline(_ZarrConnector(), cast(WritableRegistry, registry), cache_dir=tmp_path / "cache") - - assert registry.values_backend == "zarr" - - class _CountingConnector(_DemoConnector): def __init__(self) -> None: self.downloads = 0 @@ -298,15 +273,11 @@ def convert(self, raw_refs): return TimeFDataset(metadata=self.metadata()) -@pytest.mark.parametrize("pipeline", ["run", "publish"]) -def test_a_connector_without_compose_fails_before_opening_parents_or_downloading(tmp_path, monkeypatch, pipeline): +def test_a_connector_without_compose_fails_before_opening_parents_or_downloading(tmp_path, monkeypatch): monkeypatch.setattr(_ConvertOnlyChild, "download", lambda self, cache_dir: pytest.fail("downloaded")) # The parent is missing too: the check runs before the engine looks for it. with pytest.raises(TimeFValidationError, match=r"declares parents \['parent'\].*implement compose"): - if pipeline == "run": - run_pipeline(_ConvertOnlyChild(), tmp_path, cache_dir=tmp_path / "cache") - else: - publish_pipeline(_ConvertOnlyChild(), LocalRegistry(tmp_path), cache_dir=tmp_path / "cache") + run_pipeline(_ConvertOnlyChild(), tmp_path, cache_dir=tmp_path / "cache") assert not (tmp_path / "cache").exists() @@ -339,7 +310,6 @@ def compose(self, raw_refs, context): return dataset -@pytest.mark.parametrize("pipeline", ["run", "publish"]) @pytest.mark.parametrize( ("connector", "differ"), [ @@ -349,9 +319,7 @@ def compose(self, raw_refs, context): ], ids=["dropped-parent", "other-version", "dropped-parent-in-place"], ) -def test_a_dataset_with_another_card_than_the_engine_read_fails_and_stores_no_release( - tmp_path, pipeline, connector, differ -): +def test_a_dataset_with_another_card_than_the_engine_read_fails_and_stores_no_release(tmp_path, connector, differ): root = tmp_path / "root" LocalRegistry(root).put_dataset(make_dataset()) before = sorted(root.rglob("*")) @@ -359,10 +327,7 @@ def test_a_dataset_with_another_card_than_the_engine_read_fails_and_stores_no_re TimeFValidationError, match=rf"the dataset that compose\(\) returned has another card than test/child: {differ}$", ): - if pipeline == "run": - run_pipeline(connector(), root, cache_dir=tmp_path / "cache") - else: - publish_pipeline(connector(), LocalRegistry(root), cache_dir=tmp_path / "cache") + run_pipeline(connector(), root, cache_dir=tmp_path / "cache") assert sorted(root.rglob("*")) == before @@ -447,72 +412,3 @@ def record_close(reader): assert closed == [] assert closed == ["test/second", _PARENT.dataset_id] - - -class _GatedConnector(_DemoConnector): - """The demo dataset, credentialed.""" - - def metadata(self): - return make_dataset().metadata.model_copy( - update={"access": Access.CREDENTIALED, "access_url": "https://example.org/dua"} - ) - - def convert(self, raw_refs: list[str]) -> TimeFDataset: - dataset = make_dataset() - dataset._metadata = self.metadata() - return dataset - - -def test_publish_pipeline_refuses_a_card_that_is_not_open_before_downloading(tmp_path, monkeypatch): - monkeypatch.setattr(_GatedConnector, "download", lambda self, cache_dir: pytest.fail("downloaded")) - remote = RemoteRegistry( - "http://api.local", transport=httpx.MockTransport(lambda request: httpx.Response(404)), cache_dir=tmp_path - ) - - with pytest.raises(TimeNetAccessError) as caught: - publish_pipeline(_GatedConnector(), remote, cache_dir=tmp_path / "cache") - - # A card has no lineage, so its own page is the only one. - assert (caught.value.dataset, caught.value.requirements) == ( - _PARENT, - (AccessRequirement(_PARENT, Access.CREDENTIALED, "https://example.org/dua"),), - ) - # A local registry is the user's own, so it may hold the credentialed release. - monkeypatch.undo() - publish_pipeline(_GatedConnector(), LocalRegistry(tmp_path / "local"), cache_dir=tmp_path / "cache") - assert LocalRegistry(tmp_path / "local").exists(_PARENT.dataset_id, str(_PARENT.version)) - - -class _HostedRegistry: - """A registry that is not local. It serves the parents from a local registry and refuses writes.""" - - requires_open_access = True - - def __init__(self, root: Path) -> None: - self.local = LocalRegistry(root) - - def exists(self, dataset_id, version): - return False - - def open_version(self, dataset_id, version=None): - return self.local.open_version(dataset_id, version) - - def put_dataset(self, dataset, **kwargs): - pytest.fail("wrote a child whose parent is credentialed") - - -def test_publish_pipeline_refuses_an_open_child_of_a_credentialed_parent_before_the_write(tmp_path): - _store_credentialed_parent(tmp_path) - registry = cast(WritableRegistry, _HostedRegistry(tmp_path)) - - with pytest.raises(TimeNetAccessError) as caught: - publish_pipeline(_ChildConnector(), registry, cache_dir=tmp_path / "cache") - - # The card of the child is open, so the check before the download passes. Its lineage makes it - # credentialed. - child = DatasetRef(dataset_id="test/child", version=_PARENT.version) - assert (caught.value.dataset, caught.value.access, caught.value.requirements) == ( - child, - Access.CREDENTIALED, - (AccessRequirement(_PARENT, Access.CREDENTIALED, "https://example.org/dua"),), - ) diff --git a/pyproject.toml b/pyproject.toml index 10c0d2eb3..84360d028 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -67,6 +67,7 @@ select = [ "DOC", # pydoclint — args/returns match signatures "PL", # Pylint — error/warning/convention rules (opinionated metrics dropped in `ignore`) "banned-api", # flake8-tidy-imports (TID251) — the imports banned below + "banned-module-level-imports", # flake8-tidy-imports (TID253) — the lazy imports below ] ignore = [ "line-too-long", # Handled by the formatter @@ -111,10 +112,19 @@ ignore = [ "**/src/timenet/testing/*.py" = ["assert"] # The ban on importing timenet_connectors applies only to the code of the timenet package. "!packages/timenet/src/**" = ["banned-api"] +# Only the two CLIs load the CLI libraries at import. So the SDK loads the builder backend through +# its entry point without them. +"!packages/{timenet,timenet-connectors}/src/**" = ["banned-module-level-imports"] +"packages/timenet/src/timenet/{__main__.py,cli/**}" = ["banned-module-level-imports"] +"packages/timenet-connectors/src/timenet_connectors/builder/cli.py" = ["banned-module-level-imports"] [tool.ruff.lint.pydocstyle] convention = "google" +[tool.ruff.lint.flake8-tidy-imports] +# The CLI libraries, which library code imports only inside a function. +banned-module-level-imports = ["typer", "timenet.cli", "timenet_connectors.builder.cli"] + [tool.ruff.lint.flake8-tidy-imports.banned-api] # timenet-connectors depends on timenet, so timenet never imports it and works without it. "timenet_connectors".msg = "timenet must work without timenet-connectors"