Skip to content

refactor(build)!: let the connector choose the values backend #646

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
karhu merged 4 commits into from
Nov 10, 2026
Merged

refactor(build)!: let the connector choose the values backend #646

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b7a4313
refactor(build)!: remove the values backend override and move the pla…
karhu Nov 3, 2026
751e075
test(build): drop checks that only prove the values backend option is…
karhu Nov 4, 2026
be1dddc
refactor(build)!: let the connector alone choose the values backend
karhu Nov 4, 2026
a54a43b
docs: shorten the docstrings of the build groundwork
karhu Nov 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 5 additions & 16 deletions packages/timenet-connectors/src/timenet_connectors/builder/backend.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,43 +64,32 @@ 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,
root,
fetch_from=fetch_from,
build_parents=build_parents,
force=force,
values_backend=values_backend,
pinned_by=pinned_by,
)
)
101 changes: 18 additions & 83 deletions packages/timenet-connectors/src/timenet_connectors/builder/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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

Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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)
Expand All @@ -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,
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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`.

Expand Down Expand Up @@ -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.
Expand All @@ -414,25 +356,19 @@ 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():
if isinstance(target, Path):
version_dir = run_pipeline(
connector,
target,
values_backend=resolved_backend,
progress_cb=progress_cb,
force=force,
keep_cache=keep_cache,
Expand All @@ -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,
Expand Down
45 changes: 15 additions & 30 deletions packages/timenet-connectors/src/timenet_connectors/builder/env.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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())


Expand Down Expand Up @@ -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.
Expand All @@ -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:
Expand Down
Loading
Loading