Engine

timenet.engine ¶source

The build engine: drive a connector through download -> compose -> derive_schema -> store.

open_parents ¶source

open_parents(
    metadata: DatasetMetadata, registry: BaseRegistry
) -> BuildContext

Open each declared parent of a card from the registry a build writes to.

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.

Parameters:

Name	Type	Description	Default
`metadata`	`DatasetMetadata`	The child dataset card with exact parent references.	required
`registry`	`BaseRegistry`	The registry the child is written to.	required

Returns:

Type	Description
`BuildContext`	The opened readers, keyed by parent alias. Close the context after the write.

Raises:

Type	Description
`TimeNetDatasetNotFoundError`	If the registry does not hold a declared parent.

publish_pipeline ¶source

publish_pipeline(
    connector: BaseConnector,
    registry: WritableRegistry,
    *,
    cache_dir: Path | None = None,
    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.

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.

Parameters:

Name	Type	Description	Default
`connector`	`BaseConnector`	The connector to build.	required
`registry`	`WritableRegistry`	The writable registry to publish into (local, remote, or S3).	required
`cache_dir`	`Path \| None`	Directory for downloaded artifacts (defaults to `<TIMENET_CACHE>/<dataset_id>`).	`None`
`keep_cache`	`bool`	Keep the cache directory instead of removing it after publishing.	`False`
`values_backend`	`str \| None`	Storage backend for the values plane (`"parquet"` or `"zarr"`). When `None`, this function uses the connector's `values_backend`.	`None`
`progress_cb`	`Callable[[WriteProgressEvent], None] \| None`	Optional writer progress callback.	`None`
`force`	`bool`	Republish even if the version is already committed.	`False`
`pinned_by`	`Mapping[DatasetRef, PayloadSource] \| None`	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.	`None`

Returns:

Type	Description
`str`	The published version string.

Raises:

Type	Description
`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.

run_pipeline ¶source

run_pipeline(
    connector: BaseConnector,
    root: Path,
    *,
    cache_dir: Path | None = None,
    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.

This function is idempotent. If the target version is already committed, it skips the expensive download, convert, and store stages and returns the existing directory. Pass force to rebuild it. Otherwise the stages run in order: create the cache directory, open the declared parents from root, download raw references into the cache, compose them into a dataset, check that the dataset has the card that the engine read, record each declared parent in it, derive_schema, store, then delete the cache directory. Parents must already be in root; the engine never fetches or builds them. A card that declares parents needs a connector that overrides compose; otherwise the build raises before it opens a parent or downloads anything. The engine only writes local files. Publishing to a remote registry is a separate step.

Parameters:

Name	Type	Description	Default
`connector`	`BaseConnector`	The connector to build.	required
`root`	`Path`	Output root. This function writes the dataset to `<root>/<dataset_id>/<version>/`.	required
`cache_dir`	`Path \| None`	Directory for downloaded artifacts (defaults to `<TIMENET_CACHE>/<dataset_id>`).	`None`
`keep_cache`	`bool`	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.	`False`
`values_backend`	`str \| None`	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.	`None`
`progress_cb`	`Callable[[WriteProgressEvent], None] \| None`	Optional writer progress callback.	`None`
`force`	`bool`	Rebuild even if the version is already committed.	`False`
`pinned_by`	`Mapping[DatasetRef, PayloadSource] \| None`	The releases that already read files of this release, with their pins. The write publishes only if it reproduces each pinned file, and otherwise raises.	`None`

Returns:

Type	Description
`Path`	The committed version directory.

Raises:

Type	Description
`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. The build then stores no release.

store_dataset ¶source

store_dataset(
    dataset: TimeFDataset,
    root: Path,
    *,
    progress_cb: Callable[[WriteProgressEvent], None]
    | None = None,
    values_backend: str = "parquet",
    replace: bool = False,
    pinned_by: Mapping[DatasetRef, PayloadSource]
    | None = None,
) -> Path

Serialize a populated dataset to the TimeF format under root.

If the dataset has no schema, this function derives it first. It then streams the dataset through a :class:~timenet.writer.TimeFWriter. This function lives on the engine, not on :class:~timenet.connectors.BaseConnector, because it reads only dataset. This keeps the connector contract at fetch-and-convert and avoids a connector-to-writer dependency.

Parameters:

Name	Type	Description	Default
`dataset`	`TimeFDataset`	The populated dataset from `convert`.	required
`root`	`Path`	Parent directory. This function creates the version directory beneath it.	required
`progress_cb`	`Callable[[WriteProgressEvent], None] \| None`	Optional writer progress callback.	`None`
`values_backend`	`str`	Storage backend for the values plane.	`'parquet'`
`replace`	`bool`	Replace the version if it is already committed, keeping it until the new one is published.	`False`
`pinned_by`	`Mapping[DatasetRef, PayloadSource] \| None`	The releases that already read files of this release, with their pins. The writer publishes only if it reproduces each pinned file.	`None`

Returns:

Type	Description
`Path`	The committed version directory.