diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/backend.py b/packages/timenet-connectors/src/timenet_connectors/builder/backend.py index 7bc299879..bd3c9d7ab 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/backend.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/backend.py @@ -64,35 +64,25 @@ def build( # ruff: ignore[too-many-arguments, no-self-use] (protocol) root: Path, *, force: bool = False, - values_backend: str | None = None, build_parents: bool = False, fetch_from: str | BaseRegistry | None = None, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> Path: - """Build the dataset, in an isolated environment unless isolation is turned off. + """Build a dataset in an isolated environment. - ``TIMENET_ISOLATION=off`` builds in this interpreter instead, as ``--no-isolation`` does on - the CLI. That is also the recursion guard: an isolated child carries the setting, so a - connector that loads another dataset from inside one does not re-exec uv forever. The build - runs through :func:`~timenet_connectors.builder.cli.build_one`, which prepares missing parent - releases first, each the same way as this build. + Set ``TIMENET_ISOLATION=off`` to run in the current interpreter. Args: dataset_id: The dataset id. root: The output registry directory. - force: Rebuild even if the version is already built. - values_backend: Storage backend for the values plane. Defaults to the connector's value. + 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: The releases that already read files of this release, with their pins. The - build publishes the release only if it reproduces each pinned file. + pinned_by: Releases that read this dataset's files, with file pins the build must match. Returns: The committed version directory. - - Raises: - TimeFValidationError: If ``values_backend`` names a backend that TimeF does not support. - """ # ruff: ignore[docstring-extraneous-exception] - raised by build_one + """ return Path( build_one( dataset_id, @@ -100,7 +90,6 @@ def build( # ruff: ignore[too-many-arguments, no-self-use] (protocol) fetch_from=fetch_from, build_parents=build_parents, force=force, - values_backend=values_backend, pinned_by=pinned_by, ) ) diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py index b07528556..ffed1d74f 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py @@ -28,6 +28,7 @@ from timenet.cli.runner import run_cli from timenet.cli.ui import console from timenet.config import settings +from timenet.coordinator.plan import ParentStep, prepare_parents from timenet.engine import publish_pipeline, run_pipeline from timenet.errors import ( TimeFValidationError, @@ -46,10 +47,8 @@ open_writable_registry, ) from timenet.types.references import DatasetRef -from timenet.values_backends import SUPPORTED_VALUES_BACKENDS from timenet.writer.progress import ProgressStage, WriteProgressEvent from timenet_connectors.builder.env import run_isolated -from timenet_connectors.builder.plan import ParentStep, prepare_parents from timenet_connectors.discovery import connector_dir, resolve from timenet_connectors.download import DownloadProgress, ProgressCallback, progress_sink @@ -108,11 +107,6 @@ def build( keep_cache: bool = typer.Option( False, "--keep-cache", help="Keep the raw download cache after building (default: remove it)." ), - values_backend: str | None = typer.Option( - None, - "--values-backend", - help="Values-plane backend: 'parquet' or 'zarr' (default: the connector's declared backend).", - ), isolation: bool | None = typer.Option( None, "--isolation/--no-isolation", @@ -145,25 +139,14 @@ def build( the version, the build needs only the payload sources of the version. It prepares them the same way as a parent. - ``--values-backend`` overrides the values-plane storage the connector declares. This process - resolves the connector's default for an in-process build; an isolated child resolves its own, - because importing the connector there would need the dependencies it is about to install. - Raises: - BadParameter: If ``dataset_id`` has no known connector, or ``--values-backend`` names a - backend TimeF does not support. + BadParameter: If ``dataset_id`` has no known connector. TimeNetMissingParentError: If a parent is missing. The hint of the error names the flags. TimeNetFileError: If a copy of a parent in the output is damaged, or a fetched file does not match its manifest record. The hint of the error names the flags. TimeNetAccessError: If the output is remote and gets a release without access declared ``open``. """ - # Reject an unknown backend before anything expensive happens, so neither the isolated child nor - # a connector import runs for a build that cannot succeed. - try: - _check_values_backend(values_backend) - except TimeFValidationError as exc: - raise typer.BadParameter(str(exc)) from exc try: # connector_dir imports no connector. So an unknown id fails before any environment exists. connector_dir(dataset_id) @@ -179,7 +162,6 @@ def build( fetch_from=fetch_from, build_parents=build_parents, force=force, - values_backend=values_backend, keep_cache=keep_cache, isolation=isolation, narrate=True, @@ -208,50 +190,30 @@ def build_one( # ruff: ignore[too-many-arguments] fetch_from: str | BaseRegistry | None = None, build_parents: bool = False, force: bool = False, - values_backend: str | None = None, keep_cache: bool = False, isolation: bool | None = None, narrate: bool = False, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> str: - """Build one dataset: prepare its parent releases, then run its connector. - - Every build entry point calls this function. These are the ``build`` command, - :class:`~timenet_connectors.builder.backend.ConnectorBuilder`, and - :func:`timenet_connectors.build`. If ``fetch_from`` or ``build_parents`` does not obtain a - missing parent, the build fails before any download. A parent builds the same way as this - dataset. A parent never builds with ``force``. If another builder backend claims a parent, the - ``build`` method of that backend builds the parent with its pins. If ``force`` is not set and - the output already holds the version, the build prepares only the missing payload sources of - the version. In that case, the build does not run the connector of this dataset. The connector - runs in an environment built from its requirements. If isolation is off, the connector runs in - this interpreter. A local output gets a version directory. A remote output gets a published - version. + """Prepare a dataset's parent releases, then build the dataset. + + A missing parent that ``fetch_from`` or ``build_parents`` does not provide fails the build before any download. + The build reuses an existing version unless ``force`` is true. Args: dataset_id: The dataset id. - output: The output registry. This is a local directory, or a ``timenet://``, - ``http(s)://``, or ``s3://`` URL. An isolated child gets it unchanged as ``--out``. + output: The output directory or a ``timenet://``, ``http(s)://``, or ``s3://`` registry URL. fetch_from: The registry, or its URI, to copy missing parent releases from. build_parents: Build missing parent releases from their installed connectors. - force: Rebuild even if the version is already built. - values_backend: The values-plane backend, or ``None`` for the backend that the connector - declares. + force: Rebuild an existing version. keep_cache: Keep the raw download cache after building. - isolation: If true, build in an environment built from the connector's requirements. - If ``None``, use the ``TIMENET_ISOLATION`` setting. - narrate: Print status lines and download progress for an in-process build. - pinned_by: The releases that already read files of this release, with their pins. The build - publishes the release only if it reproduces each pinned file. + isolation: Use isolation if true, this interpreter if false, or ``TIMENET_ISOLATION`` if ``None``. + narrate: Print status and download progress when isolation is off. + pinned_by: Releases that read this dataset's files, with file pins the build must match. Returns: The committed version directory for a local output, or the published version. - - Raises: - TimeFValidationError: If ``values_backend`` names a backend that TimeF does not support. - This check comes first, before the plan prepares any parent. - """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_values_backend - _check_values_backend(values_backend) + """ 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") @@ -294,28 +256,11 @@ def build_parent(step: ParentStep) -> object: isolated=isolated, force=force, keep_cache=keep_cache, - values_backend=values_backend, narrate=narrate, pinned_by=pinned_by, ) -def _check_values_backend(values_backend: str | None) -> None: - """Reject the name of a values backend that TimeF does not support. - - Args: - values_backend: The name of the backend, or ``None`` for the backend that the connector - declares. ``None`` always passes. - - Raises: - TimeFValidationError: If TimeF does not support the backend. - """ - if values_backend is not None and values_backend not in SUPPORTED_VALUES_BACKENDS: - raise TimeFValidationError( - f"unknown values backend {values_backend!r}; supported: {', '.join(sorted(SUPPORTED_VALUES_BACKENDS))}" - ) - - def _in_repo(builder: BuilderBackend) -> bool: """Report whether a parent that ``builder`` claims builds through :func:`_build`. @@ -380,23 +325,20 @@ def _build( # ruff: ignore[too-many-arguments] isolated: bool, force: bool = False, keep_cache: bool = False, - values_backend: str | None = None, narrate: bool = False, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> str: - """Build one dataset into the output and return its result line. + """Build a dataset in a local directory or writable registry. Args: dataset_id: The dataset id. - target: The resolved output for an in-process build. - forwarded: The output as the caller gave it, for the ``--out`` of an isolated child. + target: The output for a build in the current interpreter. + forwarded: The output path or URL to pass to an isolated build. isolated: Build in an environment built from the connector's requirements. - force: Rebuild even if the version is already built. + force: Rebuild an existing version. keep_cache: Keep the raw download cache after building. - values_backend: The values-plane backend, or ``None`` for the backend that the connector - declares. - narrate: Print status lines and download progress for an in-process build. - pinned_by: The releases that already read files of this release, with their pins. + narrate: Print status and download progress when isolation is off. + pinned_by: Releases that read this dataset's files, with file pins the build must match. Returns: The committed version directory for a local output, or the published version. @@ -414,17 +356,12 @@ def _build( # ruff: ignore[too-many-arguments] forwarded, force=force, keep_cache=keep_cache, - values_backend=values_backend, quiet=console.quiet, pinned_by=pinned_by, ) if narrate: console.status("🔧", f"Building '{dataset_id}'…") - # Resolve the connector's declared backend here, so the in-process path and the isolated child - # agree on what "no --values-backend" means. An isolated build cannot resolve it: importing the - # connector needs the dependencies the child is about to install. connector = resolve(dataset_id)() - resolved_backend = values_backend if values_backend is not None else connector.values_backend progress_cb = _report_progress if narrate else None # A connector's downloads report through the ambient progress sink. _download_progress renders them. with _download_progress() if narrate else nullcontext(): @@ -432,7 +369,6 @@ def _build( # ruff: ignore[too-many-arguments] version_dir = run_pipeline( connector, target, - values_backend=resolved_backend, progress_cb=progress_cb, force=force, keep_cache=keep_cache, @@ -444,7 +380,6 @@ def _build( # ruff: ignore[too-many-arguments] version = publish_pipeline( connector, target, - values_backend=resolved_backend, progress_cb=progress_cb, force=force, keep_cache=keep_cache, diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/env.py b/packages/timenet-connectors/src/timenet_connectors/builder/env.py index 39f08790b..23c1e7390 100644 --- a/packages/timenet-connectors/src/timenet_connectors/builder/env.py +++ b/packages/timenet-connectors/src/timenet_connectors/builder/env.py @@ -48,12 +48,11 @@ class EnvSpec: """The interpreter the environment is built on. Always one outside a virtualenv.""" -def env_spec(dataset_id: str, *, values_backend: str | None = None) -> EnvSpec: - """Describe the environment a dataset's connector needs. +def env_spec(dataset_id: str) -> EnvSpec: + """Describe the dependencies and interpreter for a connector's build. Args: dataset_id: The dataset id. - values_backend: Selected backend, or ``None`` for a default resolved by the child. Returns: The environment specification. @@ -63,11 +62,10 @@ def env_spec(dataset_id: str, *, values_backend: str | None = None) -> EnvSpec: TimeNetBuildError: If a base distribution is not installed. """ # ruff: ignore[docstring-extraneous-exception] (raised by requirements_for and _base_args, not directly here) base = _base_args() - # The parent cannot import the connector before its dependencies exist. When its default is - # unknown, include the optional backend too. An explicit Parquet build needs no Zarr dependency. - # Keep the parent's source/version pin; the additional requirement only requests its extra. - if values_backend != "parquet": - base += ("--with", f"timenet[zarr]=={Distribution.from_name('timenet').version}") + # The parent cannot read the connector's backend before the environment exists, so the + # environment always includes the optional Zarr backend. Keep the parent's source/version pin; + # the additional requirement only requests its extra. + base += ("--with", f"timenet[zarr]=={Distribution.from_name('timenet').version}") return EnvSpec(base=base, requirements=requirements_for(dataset_id), python=_base_interpreter()) @@ -107,37 +105,26 @@ def run_isolated( # ruff: ignore[too-many-arguments] *, force: bool = False, keep_cache: bool = False, - values_backend: str | None = None, quiet: bool = False, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> str: - """Build a dataset in its own environment and return what the child printed on stdout. + """Build a dataset in a separate environment with its connector's dependencies. - The child's stderr is streamed live (a long download still shows progress) and also captured, so - a failed build's message carries the child's own error. Its stdout is captured, because the last - line is the build's result: the committed version directory for a local ``out``, or the published - version for a remote one. ``out`` is forwarded to the child's ``--out`` unchanged, so it publishes - or writes exactly as a direct build would. + The build streams status and errors to stderr. Args: dataset_id: The dataset id. - out: The output target, forwarded to the child's ``--out``: a local directory or a - ``timenet://`` / ``http(s)://`` / ``s3://`` URL. - force: Rebuild even if the version is already built. + out: The output directory or a ``timenet://``, ``http(s)://``, or ``s3://`` registry URL. + force: Rebuild an existing version. keep_cache: Keep the raw download cache after building. - values_backend: Values-plane backend to forward to the child (``"parquet"`` or - ``"zarr"``). ``None`` forwards nothing, leaving the child on the connector's - declared backend. - quiet: Suppress the child's status output, as ``--quiet`` does in this process. - pinned_by: The releases that already read files of the dataset, with their pins. The child - receives them in a file through the hidden ``--pinned-by`` option, and publishes only - if it reproduces each pinned file. + quiet: Suppress the build's status output. + pinned_by: Releases that read this dataset's files, with file pins the build must match. Returns: - The child's last non-empty stdout line: a version directory or a published version. + The version directory for a local output, or the published version for a remote output. Raises: - TimeNetBuildError: If the child failed, or printed nothing. + TimeNetBuildError: If the build fails or prints no result. LookupError: If no connector exists for the id. """ # ruff: ignore[docstring-extraneous-exception] (raised by env_spec, not directly here) # --quiet belongs to timenet-build, not to build, so it goes before the subcommand. @@ -146,9 +133,7 @@ def run_isolated( # ruff: ignore[too-many-arguments] argv.append("--force") if keep_cache: argv.append("--keep-cache") - if values_backend is not None: - argv += ["--values-backend", values_backend] - spec = env_spec(dataset_id, values_backend=values_backend) + spec = env_spec(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: diff --git a/packages/timenet-connectors/tests/test_build_backend.py b/packages/timenet-connectors/tests/test_build_backend.py index 53a9679e1..63ec0616e 100644 --- a/packages/timenet-connectors/tests/test_build_backend.py +++ b/packages/timenet-connectors/tests/test_build_backend.py @@ -9,6 +9,7 @@ from timenet.reader import TimeFReader from timenet.registry import open_local_version from timenet.types.references import DatasetRef +from timenet.values_backends import ValuesBackend from timenet_connectors.builder import backend as backend_module, cli as cli_module from timenet_connectors.builder.backend import ConnectorBuilder @@ -42,21 +43,20 @@ def test_a_card_that_cannot_be_read_raises_and_only_a_missing_connector_gives_no ConnectorBuilder().declared_version("test/broken") -@pytest.mark.parametrize("values_backend", [None, "zarr", "parquet"]) -def test_build_runs_in_an_isolated_environment_by_default(tmp_path, monkeypatch, values_backend): +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, values_backend=None, **_): - calls.append((dataset_id, root, force, values_backend)) + 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, values_backend=values_backend) + 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, values_backend)] + assert calls == [("timenet/hello-world", tmp_path, True)] def test_build_hands_its_pins_to_the_build(tmp_path, monkeypatch): @@ -93,7 +93,7 @@ def test_build_runs_in_process_when_isolation_is_off(tmp_path, monkeypatch): 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") - monkeypatch.setattr(connector_cls, "values_backend", "zarr") + 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") @@ -102,20 +102,3 @@ def test_build_in_process_round_trips_the_connector_default_zarr_backend(tmp_pat dataset = reader.read() assert len(dataset.records) == 3 assert len(dataset.records[0].signals[0].to_numpy()) == 16 - - -def test_build_in_process_uses_the_connector_default_values_backend(tmp_path, monkeypatch): - class Connector: - values_backend = "zarr" - - captured = {} - monkeypatch.setenv("TIMENET_ISOLATION", "off") - monkeypatch.setattr(cli_module, "resolve", lambda dataset_id: Connector) - - def pipeline(connector, root, *, force, values_backend, **_): - captured["backend"] = values_backend - return Path(root) / "1.0.0" - - monkeypatch.setattr(cli_module, "run_pipeline", pipeline) - ConnectorBuilder().build("timenet/hello-world", tmp_path) - assert captured["backend"] == "zarr" diff --git a/packages/timenet-connectors/tests/test_build_cli.py b/packages/timenet-connectors/tests/test_build_cli.py index bdda85475..495e6b0e7 100644 --- a/packages/timenet-connectors/tests/test_build_cli.py +++ b/packages/timenet-connectors/tests/test_build_cli.py @@ -1,4 +1,3 @@ -from importlib.util import find_spec import os from pathlib import Path import subprocess @@ -7,7 +6,6 @@ import pytest from typer.testing import CliRunner -from timenet.errors import TimeFValidationError from timenet.manifest import Manifest from timenet.reader import TimeFReader from timenet.registry import open_local_version @@ -22,7 +20,7 @@ def test_build_runs_isolated_by_default(monkeypatch, tmp_path): captured = {} def _fake_isolated( # ruff: ignore[too-many-arguments] (mirrors run_isolated's full keyword surface) - dataset_id, root, *, force=False, keep_cache=False, quiet=False, values_backend=None, pinned_by=None + dataset_id, root, *, force=False, keep_cache=False, quiet=False, pinned_by=None ): captured["dataset_id"] = dataset_id captured["root"] = root @@ -70,95 +68,12 @@ def _fake_pipeline(connector, root, **kwargs): assert called["isolated"] is False -def test_build_without_a_values_backend_uses_the_connector_default(monkeypatch, tmp_path): - captured = {} - instances = [] - - class _Connector: - values_backend = "zarr" - - def __init__(self): - instances.append(self) - - monkeypatch.setenv("TIMENET_ISOLATION", "off") - monkeypatch.setattr(cli_module, "resolve", lambda dataset_id: _Connector) - - def _fake_pipeline(connector, root, **kwargs): - assert connector is instances[0] - captured["values_backend"] = kwargs["values_backend"] - return Path(root) / "timenet/hello-world/1.0.0" - - monkeypatch.setattr(cli_module, "run_pipeline", _fake_pipeline) - result = runner.invoke(app, ["build", "timenet/hello-world", "--out", str(tmp_path)]) - - assert result.exit_code == 0 - assert captured["values_backend"] == "zarr" - assert len(instances) == 1 - - -def test_build_forwards_an_explicit_values_backend_in_process(monkeypatch, tmp_path): - captured = {} - monkeypatch.setenv("TIMENET_ISOLATION", "off") - - def _fake_pipeline(connector, root, **kwargs): - captured["values_backend"] = kwargs["values_backend"] - return Path(root) / "timenet/hello-world/1.0.0" - - monkeypatch.setattr(cli_module, "run_pipeline", _fake_pipeline) - result = runner.invoke( - app, - ["build", "timenet/hello-world", "--out", str(tmp_path), "--values-backend", "zarr"], - ) - - assert result.exit_code == 0 - assert captured["values_backend"] == "zarr" - - -def test_build_forwards_an_explicit_values_backend_to_an_isolated_child(monkeypatch, tmp_path): - captured = {} - - def _fake_isolated( # ruff: ignore[too-many-arguments] (mirrors run_isolated's full keyword surface) - dataset_id, root, *, force=False, keep_cache=False, quiet=False, values_backend=None, pinned_by=None - ): - captured["values_backend"] = values_backend - return Path(root) / "timenet/hello-world/1.0.0" - - monkeypatch.setattr(cli_module, "run_isolated", _fake_isolated) - result = runner.invoke( - app, - ["build", "timenet/hello-world", "--out", str(tmp_path), "--values-backend", "zarr"], - ) - - assert result.exit_code == 0 - assert captured["values_backend"] == "zarr" - - -def test_build_rejects_an_unknown_values_backend(tmp_path): - result = runner.invoke( - app, - ["build", "timenet/hello-world", "--out", str(tmp_path), "--values-backend", "unknown"], - ) - - assert result.exit_code != 0 - assert "values backend" in result.output.lower() - - -def test_build_one_rejects_an_unknown_values_backend_before_it_prepares_parents(tmp_path, monkeypatch): - def prepare_parents(*args, **kwargs): - pytest.fail("build_one prepared the parents") - - monkeypatch.setattr(cli_module, "prepare_parents", prepare_parents) - - with pytest.raises(TimeFValidationError, match=r"^unknown values backend 'unknown'; supported: parquet, zarr$"): - cli_module.build_one("timenet/hello-world", tmp_path, values_backend="unknown") - - def test_build_isolates_when_the_flag_overrides_the_setting(monkeypatch, tmp_path): called = {"isolated": False} monkeypatch.setenv("TIMENET_ISOLATION", "off") def _fake_isolated( # ruff: ignore[too-many-arguments] (mirrors run_isolated's full keyword surface) - dataset_id, root, *, force=False, keep_cache=False, quiet=False, values_backend=None, pinned_by=None + dataset_id, root, *, force=False, keep_cache=False, quiet=False, pinned_by=None ): called["isolated"] = True return Path(root) / "timenet/hello-world/1.0.0" @@ -197,7 +112,7 @@ def test_build_forwards_quiet_to_the_isolated_child(monkeypatch, tmp_path): monkeypatch.setattr(cli_module.console, "quiet", False) def _fake_isolated( # ruff: ignore[too-many-arguments] (mirrors run_isolated's full keyword surface) - dataset_id, root, *, force=False, keep_cache=False, quiet=False, values_backend=None, pinned_by=None + dataset_id, root, *, force=False, keep_cache=False, quiet=False, pinned_by=None ): captured["quiet"] = quiet return Path(root) / "timenet/hello-world/1.0.0" @@ -210,8 +125,7 @@ def _fake_isolated( # ruff: ignore[too-many-arguments] (mirrors run_isolated's @pytest.mark.slow -@pytest.mark.parametrize("backend", [None, "zarr", "parquet"]) -def test_isolated_build_prints_the_version_dir_and_narrates_once(tmp_path, backend): +def test_isolated_build_prints_the_version_dir_and_narrates_once(tmp_path): # run_isolated reads the child's last non-empty stdout line as the version directory, so the # real isolated CLI must keep stdout clean and end on that path. Drive it as a subprocess (not # CliRunner) so the child's inherited stderr is captured here too, where a double-narration @@ -220,9 +134,8 @@ def test_isolated_build_prints_the_version_dir_and_narrates_once(tmp_path, backe env = {**os.environ, "TIMENET_HOME": str(tmp_path / "home")} env.pop("TIMENET_ISOLATION", None) # default is "on"; make sure nothing forces it off run_cli = [sys.executable, "-c", "from timenet_connectors.builder.cli import main; main()"] - backend_args = ["--values-backend", backend] if backend else [] result = subprocess.run( - [*run_cli, "build", "timenet/hello-world", "--out", str(registry), *backend_args], + [*run_cli, "build", "timenet/hello-world", "--out", str(registry)], env=env, cwd=tmp_path, # away from the repo tree so the CLI does not read the repo .env capture_output=True, @@ -235,13 +148,10 @@ def test_isolated_build_prints_the_version_dir_and_narrates_once(tmp_path, backe stdout_lines = [line.strip() for line in result.stdout.splitlines() if line.strip()] assert stdout_lines[-1] == str(version_dir) assert (version_dir / "manifest.json").is_file() - assert Manifest.model_validate_json((version_dir / "manifest.json").read_text()).values_backend == ( - backend or "parquet" - ) - if backend != "zarr" or find_spec("zarr") is not None: - with TimeFReader(open_local_version(version_dir)) as reader: - dataset = reader.read() - assert len(dataset.records[0].signals[0].to_numpy()) == 16 + assert Manifest.model_validate_json((version_dir / "manifest.json").read_text()).values_backend == "parquet" + with TimeFReader(open_local_version(version_dir)) as reader: + dataset = reader.read() + assert len(dataset.records[0].signals[0].to_numpy()) == 16 # The isolated child is the sole narrator: each status line appears exactly once. assert result.stderr.count("🔧 Building 'timenet/hello-world'…") == 1 diff --git a/packages/timenet-connectors/tests/test_build_env.py b/packages/timenet-connectors/tests/test_build_env.py index de57105d5..014ed0d69 100644 --- a/packages/timenet-connectors/tests/test_build_env.py +++ b/packages/timenet-connectors/tests/test_build_env.py @@ -104,9 +104,8 @@ def test_base_args_pins_a_version_for_an_index_install(monkeypatch): assert _base_args_for(monkeypatch, _FakeDist(version="9.9.9", direct_url=None)) == ("--with", "timenet==9.9.9") -@pytest.mark.parametrize("backend", [None, "zarr", "parquet"]) @pytest.mark.parametrize("editable", [False, True]) -def test_isolated_environment_installs_backend_dependencies(monkeypatch, tmp_path, backend, editable): +def test_isolated_environment_installs_backend_dependencies(monkeypatch, tmp_path, editable): captured = {} direct_url = json.dumps({"url": tmp_path.as_uri(), "dir_info": {"editable": True}}) if editable else None monkeypatch.setattr( @@ -118,9 +117,9 @@ def run(command, env): return str(tmp_path), "", 0 monkeypatch.setattr(env_module, "_run_build", run) - run_isolated("timenet/hello-world", tmp_path, values_backend=backend) + run_isolated("timenet/hello-world", tmp_path) dependencies = captured["command"][: captured["command"].index("timenet-build")] - assert ("timenet[zarr]==9.9.9" in dependencies) == (backend != "parquet") + assert "timenet[zarr]==9.9.9" in dependencies def test_run_isolated_disables_isolation_in_the_child(monkeypatch, tmp_path): @@ -178,33 +177,6 @@ def _fake(command, env): assert "--keep-cache" in captured["command"] -def test_run_isolated_forwards_an_explicit_values_backend(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, values_backend="zarr") - - command = captured["command"] - assert command[command.index("--values-backend") + 1] == "zarr" - - -def test_run_isolated_omits_values_backend_without_an_override(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) - - assert "--values-backend" not in captured["command"] - - def test_run_isolated_forwards_quiet_ahead_of_the_subcommand(monkeypatch, tmp_path): captured = {} diff --git a/packages/timenet-connectors/tests/test_build_plan.py b/packages/timenet-connectors/tests/test_build_plan.py index 7a9ae47fe..5da7d7c06 100644 --- a/packages/timenet-connectors/tests/test_build_plan.py +++ b/packages/timenet-connectors/tests/test_build_plan.py @@ -7,6 +7,7 @@ from timenet import builders from timenet.connectors import BaseConnector +from timenet.coordinator.plan import ParentStep, plan_parents, prepare_parents from timenet.dataset import Record, RegularAxis, Signal, Source, TimeFDataset from timenet.engine import open_parents, run_pipeline from timenet.errors import ( @@ -27,12 +28,12 @@ from timenet.types.parent_dataset_card import ParentDatasetCard from timenet.types.reference_keys import ParentAlias, SourceKey from timenet.types.references import DatasetRef +from timenet.values_backends import ValuesBackend import timenet_connectors from timenet_connectors import api as api_module from timenet_connectors.builder import backend as backend_module, cli as cli_module, env as env_module from timenet_connectors.builder.backend import ConnectorBuilder from timenet_connectors.builder.cli import app -from timenet_connectors.builder.plan import ParentStep, plan_parents, prepare_parents GRAND = DatasetRef(dataset_id=make_dataset().metadata.dataset_id, version=make_dataset().metadata.dataset_version) @@ -295,9 +296,10 @@ def test_a_missing_payload_source_names_the_release_that_reads_it(tmp_path, monk assert str(caught.value).startswith("payload source timenet/hello-world@1.0.0 of test/parent@1.0.0 is not in") -@pytest.mark.parametrize("backend", ["parquet", "zarr"]) +@pytest.mark.parametrize("backend", [ValuesBackend.PARQUET, ValuesBackend.ZARR]) def test_a_pinned_source_is_rebuilt_in_staging_and_published_only_if_it_matches(tmp_path, monkeypatch, backend): _install_cards(monkeypatch, child={"parent": PARENT}, parent={"parent": GRAND}) + monkeypatch.setattr(_Grand, "values_backend", backend) output = LocalRegistry(tmp_path / "output") _parent_without_its_source(output) pins = output.get_manifest(PARENT.dataset_id).payload_sources[SourceKey(str(GRAND))] @@ -307,9 +309,9 @@ def test_a_pinned_source_is_rebuilt_in_staging_and_published_only_if_it_matches( assert steps == [ParentStep(GRAND, "build", pinned_by={PARENT: pins})] def build(step): - run_pipeline(_Grand(), output.root, values_backend=backend, pinned_by=step.pinned_by) + run_pipeline(_Grand(), output.root, pinned_by=step.pinned_by) - if backend == "parquet": + if backend == ValuesBackend.PARQUET: prepare_parents(CHILD.dataset_id, output, build=build, build_parents=True) assert output.exists(GRAND.dataset_id, str(GRAND.version)) return diff --git a/packages/timenet/src/timenet/connectors/base.py b/packages/timenet/src/timenet/connectors/base.py index 2c0951c8e..4d265fcbe 100644 --- a/packages/timenet/src/timenet/connectors/base.py +++ b/packages/timenet/src/timenet/connectors/base.py @@ -14,6 +14,7 @@ from timenet.dataset import TimeFDataset from timenet.types import DatasetMetadata +from timenet.values_backends import ValuesBackend if TYPE_CHECKING: @@ -38,8 +39,8 @@ class BaseConnector(ABC, Generic[TRaw]): """Optional explicit path to the dataset card YAML. When ``None`` (the default), the connector reads the card from ``dataset.yaml`` in its own folder.""" - values_backend: str = "parquet" - """Default storage backend for this connector's values plane.""" + values_backend: ValuesBackend = ValuesBackend.PARQUET + """Storage backend for the dataset's time-series values.""" def __new__(cls, *args: object, **kwargs: object) -> Self: """Create a connector of a class that implements its hooks. diff --git a/packages/timenet/src/timenet/coordinator/__init__.py b/packages/timenet/src/timenet/coordinator/__init__.py new file mode 100644 index 000000000..7e1691695 --- /dev/null +++ b/packages/timenet/src/timenet/coordinator/__init__.py @@ -0,0 +1 @@ +"""Coordinate dataset builds and reads.""" diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/plan.py b/packages/timenet/src/timenet/coordinator/plan.py similarity index 100% rename from packages/timenet-connectors/src/timenet_connectors/builder/plan.py rename to packages/timenet/src/timenet/coordinator/plan.py diff --git a/packages/timenet/src/timenet/engine/engine.py b/packages/timenet/src/timenet/engine/engine.py index 375420413..c7a392b08 100644 --- a/packages/timenet/src/timenet/engine/engine.py +++ b/packages/timenet/src/timenet/engine/engine.py @@ -21,7 +21,7 @@ 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 +from timenet.values_backends import SUPPORTED_VALUES_BACKENDS, ValuesBackend from timenet.writer import TimeFWriter, WriteProgressEvent @@ -33,7 +33,6 @@ def run_pipeline( # ruff: ignore[too-many-arguments] keep_cache: bool = False, progress_cb: Callable[[WriteProgressEvent], None] | None = None, force: bool = False, - values_backend: str | None = None, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> Path: """Run one connector through the full build pipeline and return the version directory. @@ -56,9 +55,6 @@ def run_pipeline( # ruff: ignore[too-many-arguments] keep_cache: Keep the cache directory instead of removing it once the dataset is stored. Conversion is the only stage that needs the raw sources, so removing them frees disk after a successful build. The sources re-download on the next run. - values_backend: Storage backend for the values plane (``"parquet"`` or ``"zarr"``). When - ``None``, this function uses the connector's ``values_backend``, so a connector that - needs Zarr declares it once on the class. progress_cb: Optional writer progress callback. force: Rebuild even if the version is already committed. pinned_by: The releases that already read files of this release, with their pins. The write @@ -72,7 +68,7 @@ def run_pipeline( # ruff: ignore[too-many-arguments] ``compose``, or the dataset that ``compose`` returns has another card than the one that the engine read. The build then stores no release. """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_composes and _compose - resolved_backend = _resolve_values_backend(connector, values_backend) + values_backend = _check_values_backend(connector) # Read only the connector's dataset.yaml card, a tiny local file, not the dataset itself. This lets # us resolve the version directory and skip the expensive download and convert when it exists. metadata = connector.metadata() @@ -95,7 +91,7 @@ def run_pipeline( # ruff: ignore[too-many-arguments] store_dataset( dataset, root, - values_backend=resolved_backend, + values_backend=values_backend, progress_cb=progress_cb, replace=force, pinned_by=pinned_by, @@ -115,43 +111,30 @@ def publish_pipeline( # ruff: ignore[too-many-arguments] keep_cache: bool = False, progress_cb: Callable[[WriteProgressEvent], None] | None = None, force: bool = False, - values_backend: str | None = None, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> str: - """Run one connector and publish the result through a writable registry. + """Build and publish a dataset to a writable registry. - Like :func:`run_pipeline`, but the converted dataset is handed to ``registry.store`` instead of - written to a local directory, so it also targets a remote or S3 registry. It is idempotent: an - already-committed version skips the ``download`` and ``convert`` stages and returns unless ``force``. - Like :func:`run_pipeline`, it checks after ``compose`` that the dataset has the card that the - engine read. - A registry other than a local one holds open releases only, so a card that is not open fails - before any download. The composed dataset can be a child whose card says open while a parent is - not open. So the engine checks the dataset again with its lineage before the write. + The build reuses an existing version unless ``force`` is true. + All declared parents must already exist in the registry. Args: connector: The connector to build. - registry: The writable registry to publish into (local, remote, or S3). - cache_dir: Directory for downloaded artifacts (defaults to ``/``). - keep_cache: Keep the cache directory instead of removing it after publishing. - values_backend: Storage backend for the values plane (``"parquet"`` or ``"zarr"``). When - ``None``, this function uses the connector's ``values_backend``. + registry: The local, remote, or S3 registry to publish to. + cache_dir: Download directory to keep, or ``None`` for ``/``. + keep_cache: Keep the default download cache after publishing. progress_cb: Optional writer progress callback. - force: Republish even if the version is already committed. - pinned_by: The releases that already read files of this release, with their pins. The store - publishes only if it reproduces each pinned file, and otherwise raises. + force: Republish an existing version. + pinned_by: Releases that read this dataset's files, with file pins the build must match. Returns: The published version string. Raises: - TimeFValidationError: If the card declares parents and the connector does not override - ``compose``, or the dataset that ``compose`` returns has another card than the one that - the engine read. Nothing is published then. - TimeNetAccessError: If the registry is not local, and the card, or the composed dataset with - its lineage, is not open. + TimeFValidationError: If a card with parents has no ``compose`` override, or ``compose`` changes the card. + TimeNetAccessError: If a release or ancestor lacks open access for a nonlocal registry. """ # ruff: ignore[docstring-extraneous-exception] - raised by _check_composes, _compose, and require_open_access - resolved_backend = _resolve_values_backend(connector, values_backend) + values_backend = _check_values_backend(connector) metadata = connector.metadata() dataset_id = metadata.dataset_id version = str(metadata.dataset_version) @@ -172,7 +155,7 @@ def publish_pipeline( # ruff: ignore[too-many-arguments] # Pass pins only when there are any, so a registry written against the earlier store() # signature keeps working for every other build. pinned = {"pinned_by": pinned_by} if pinned_by else {} - registry.store(dataset, force=force, values_backend=resolved_backend, progress_cb=progress_cb, **pinned) + registry.store(dataset, force=force, values_backend=values_backend, progress_cb=progress_cb, **pinned) if not keep_cache and cache_dir is None and cache.is_dir(): shutil.rmtree(cache) return version @@ -256,21 +239,21 @@ def _compose(connector: BaseConnector, metadata: DatasetMetadata, cache: Path, c return dataset -def _resolve_values_backend(connector: BaseConnector, override: str | None) -> str: - """Resolve and validate the backend before a build changes any files. +def _check_values_backend(connector: BaseConnector) -> ValuesBackend: + """Validate and return a connector's values backend. Returns: - The supported backend selected by the override or connector default. + The connector's supported values backend. Raises: - TimeFValidationError: If the selected backend is unknown. + TimeFValidationError: If the connector's values backend is unknown. """ - backend = connector.values_backend if override is None else override + backend = connector.values_backend if backend not in SUPPORTED_VALUES_BACKENDS: raise TimeFValidationError( f"unknown values_backend {backend!r}; supported: {', '.join(sorted(SUPPORTED_VALUES_BACKENDS))}" ) - return backend + return ValuesBackend(backend) def store_dataset( # ruff: ignore[too-many-arguments] @@ -278,7 +261,7 @@ def store_dataset( # ruff: ignore[too-many-arguments] root: Path, *, progress_cb: Callable[[WriteProgressEvent], None] | None = None, - values_backend: str = "parquet", + values_backend: ValuesBackend = ValuesBackend.PARQUET, replace: bool = False, pinned_by: Mapping[DatasetRef, PayloadSource] | None = None, ) -> Path: diff --git a/packages/timenet/tests/coordinator/test_plan_imports.py b/packages/timenet/tests/coordinator/test_plan_imports.py new file mode 100644 index 000000000..c0f08c575 --- /dev/null +++ b/packages/timenet/tests/coordinator/test_plan_imports.py @@ -0,0 +1,7 @@ +import subprocess +import sys + + +def test_the_planner_imports_nothing_from_timenet_connectors(): + code = "import sys, timenet.coordinator.plan; assert 'timenet_connectors' not in sys.modules" + subprocess.run([sys.executable, "-c", code], check=True) diff --git a/packages/timenet/tests/test_engine.py b/packages/timenet/tests/test_engine.py index 571e9a87c..d0ab1dc58 100644 --- a/packages/timenet/tests/test_engine.py +++ b/packages/timenet/tests/test_engine.py @@ -19,6 +19,7 @@ from timenet.types.access import AccessRequirement, effective_access from timenet.types.reference_keys import ParentAlias from timenet.types.references import DatasetRef +from timenet.values_backends import ValuesBackend from timenet.writer import TimeFWriter @@ -88,22 +89,16 @@ def test_run_pipeline_end_to_end(tmp_path): def test_connector_defaults_to_the_parquet_values_backend(): - assert _DemoConnector().values_backend == "parquet" + assert _DemoConnector().values_backend == ValuesBackend.PARQUET def test_run_pipeline_writes_the_requested_values_backend(tmp_path): pytest.importorskip("zarr") class _ZarrConnector(_DemoConnector): - values_backend = "zarr" - - connector = _ZarrConnector() - version_dir = run_pipeline( - connector, - tmp_path, - cache_dir=tmp_path / "cache", - values_backend=connector.values_backend, - ) + values_backend = ValuesBackend.ZARR + + version_dir = run_pipeline(_ZarrConnector(), tmp_path, cache_dir=tmp_path / "cache") manifest = Manifest.model_validate_json((version_dir / "manifest.json").read_text()) assert manifest.values_backend == "zarr" @@ -121,16 +116,10 @@ def store(self, dataset, *, force: bool, values_backend: str, progress_cb) -> No self.values_backend = values_backend class _ZarrConnector(_DemoConnector): - values_backend = "zarr" + values_backend = ValuesBackend.ZARR - connector = _ZarrConnector() registry = _Registry() - publish_pipeline( - connector, - cast(WritableRegistry, registry), - cache_dir=tmp_path / "cache", - values_backend=connector.values_backend, - ) + publish_pipeline(_ZarrConnector(), cast(WritableRegistry, registry), cache_dir=tmp_path / "cache") assert registry.values_backend == "zarr" @@ -176,21 +165,13 @@ def fail(self): assert [path.name for path in version_dir.parent.iterdir()] == [version_dir.name] -@pytest.mark.parametrize("use_default", [False, True]) -def test_invalid_backend_preserves_committed_version(tmp_path, use_default): +def test_invalid_backend_preserves_committed_version(tmp_path): connector = _CountingConnector() version_dir = run_pipeline(connector, tmp_path, cache_dir=tmp_path / "cache") before = {p.relative_to(version_dir): p.read_bytes() for p in version_dir.rglob("*") if p.is_file()} - if use_default: - connector.values_backend = "invalid" + connector.values_backend = cast("ValuesBackend", "invalid") with pytest.raises(TimeFValidationError, match="backend"): - run_pipeline( - connector, - tmp_path, - cache_dir=tmp_path / "cache", - force=True, - values_backend=None if use_default else "invalid", - ) + run_pipeline(connector, tmp_path, cache_dir=tmp_path / "cache", force=True) assert {p.relative_to(version_dir): p.read_bytes() for p in version_dir.rglob("*") if p.is_file()} == before assert connector.downloads == 1 diff --git a/scripts/gen_api_docs.py b/scripts/gen_api_docs.py index 40fb624d0..127eb1cad 100644 --- a/scripts/gen_api_docs.py +++ b/scripts/gen_api_docs.py @@ -40,6 +40,7 @@ EXCLUDE = { "timenet.cli", # console-script plumbing (Typer app), not a code API "timenet.cache", # cache-inspection helper + "timenet.coordinator", # internal planner, not a user API "timenet.torch", # optional torch adapter (covered in the guide) "timenet.schemas", # packaged JSON-Schema data, not code "timenet.testing", # test-support helpers shipped in src diff --git a/zensical.toml b/zensical.toml index ccf00701b..283c61573 100644 --- a/zensical.toml +++ b/zensical.toml @@ -22,7 +22,7 @@ nav = [ { "Datasets" = [{ "Use datasets" = [{ "The client (SDK)" = "client.md" }, { "Registries" = "registry.md" }, { "Load into your stack" = "usage.md" }] }, { "Build datasets" = [{ "Connectors" = "connectors.md" }, { "Build & publish" = "build.md" }] }, { "Catalog" = [{ "Datasets" = "catalog/datasets.md" }, { "Benchmarks" = "catalog/benchmarks.md" }, { "Licensing" = "catalog/licensing.md" }] }] }, { "CLI" = [{ "Overview" = "cli/index.md" }, { "timenet (consumer)" = "cli/timenet.md" }, { "timenet-build (producer)" = "cli/build.md" }] }, # >>> api-nav: generated by scripts/gen_api_docs.py — do not edit >>> - { "API Reference" = [{ "Overview" = "api/index.md" }, { "timenet.builders" = "api/builders.md" }, { "timenet.client" = "api/client.md" }, { "timenet.config" = "api/config.md" }, { "timenet.connectors" = "api/connectors.md" }, { "timenet.dataset" = "api/dataset.md" }, { "timenet.engine" = "api/engine.md" }, { "timenet.errors" = "api/errors.md" }, { "timenet.format" = "api/format.md" }, { "timenet.json" = "api/json.md" }, { "timenet.manifest" = "api/manifest.md" }, { "timenet.provenance" = "api/provenance.md" }, { "timenet.reader" = "api/reader.md" }, { "timenet.registry" = "api/registry.md" }, { "timenet.types" = "api/types.md" }, { "timenet.values_backends" = "api/values_backends.md" }, { "timenet.writer" = "api/writer.md" }, { "timenet_connectors.api" = "api/api.md" }, { "timenet_connectors.bases" = "api/bases.md" }, { "timenet_connectors.discovery" = "api/discovery.md" }, { "timenet_connectors.download" = "api/download.md" }, { "timenet_connectors.sources" = "api/sources.md" }, { "timenet_connectors.splitting" = "api/splitting.md" }, { "timenet_connectors.time_axes" = "api/time_axes.md" }] }, + { "API Reference" = [{ "Overview" = "api/index.md" }, { "timenet.builders" = "api/builders.md" }, { "timenet.client" = "api/client.md" }, { "timenet.config" = "api/config.md" }, { "timenet.connectors" = "api/connectors.md" }, { "timenet.dataset" = "api/dataset.md" }, { "timenet.engine" = "api/engine.md" }, { "timenet.errors" = "api/errors.md" }, { "timenet.format" = "api/format.md" }, { "timenet.json" = "api/json.md" }, { "timenet.manifest" = "api/manifest.md" }, { "timenet.provenance" = "api/provenance.md" }, { "timenet.reader" = "api/reader.md" }, { "timenet.registry" = "api/registry.md" }, { "timenet.storage" = "api/storage.md" }, { "timenet.types" = "api/types.md" }, { "timenet.values_backends" = "api/values_backends.md" }, { "timenet.writer" = "api/writer.md" }, { "timenet_connectors.api" = "api/api.md" }, { "timenet_connectors.bases" = "api/bases.md" }, { "timenet_connectors.discovery" = "api/discovery.md" }, { "timenet_connectors.download" = "api/download.md" }, { "timenet_connectors.sources" = "api/sources.md" }, { "timenet_connectors.splitting" = "api/splitting.md" }, { "timenet_connectors.time_axes" = "api/time_axes.md" }] }, # <<< api-nav <<< ]