mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-10 07:18:27 +01:00
skshetry
34d80fbfce
* add support for tracking remote dataset This adds support for virtually tracking a dataset such as dvcx dataset, dataset from remote dvc/git registries, and cloud-versioned remotes. This PR introduces two different commands under `ds` namespace: `add` and `update`. `dvc ds add` command adds the dataset to `dvc.yaml` and tracks the sources in `dvc.lock` file. Similarly, `dvc ds update` updates the sources in `dvc.lock` file. Example Usage: ```console dvc ds add --name dogs --url dvcx://dogs # tracking dvcx dataset # tracking dvc/git registries dvc ds add --name example --url dvc://[email protected]/iterative/example.git # cloud-versioned-remotes dvc ds add --name versioning --url s3://cloud-versioning-demo ``` To update, specify the name. ```console dvc ds update <name> ``` `dvc ds add` freezes (in the traditional sense of `dvc import/import-url`). It keeps a "specification" of the dataset in `dvc.yaml` and also freezes information about the dataset in the `dvc.lock` file. They are kept inside `datasets` section in both `dvc.yaml` and `dvc.lock` files. This metadata is used in the pipelines. You can add a dependency to your stage using `ds://` scheme, followed by the name of the dataset in `dvc.yaml` file. As it is used in pipelines, the `name` of the dataset has to be unique in the repository. Different dataset of same names are not allowed. On `dvc repro`, `dvc` copies the frozen information about the particular dataset into the `deps` field for the stage in `dvc.lock`. On successive invocation, `dvc` will compare the information from the deps field of the lock with the frozen information in the `datasets` section and decides whether to rerun or not. As the dataset is frozen, `dvc repro` won't rerun until the dataset is updated via `dvc update`. Here are some examples for how `dvc.yaml` looks like: ```yaml datasets: - name: dogs url: dvcx://dogs type: dvcx - name: example-get-started url: [email protected]:iterative/example-get-started.git type: dvc path: path - name: dogs2 url: dvcx://dogs@v2 type: dvcx - name: cloud-versioning-demo url: s3://cloud-versioning-demo type: url ``` ```yaml schema: '2.0' datasets: - name: dogs url: dvcx://dogs type: dvcx version: 3 created_at: '2023-12-11T10:32:05.942708+00:00' - name: example-get-started url: [email protected]:iterative/example-get-started.git type: dvc path: path rev_lock: df75c16ef61f0772d6e4bb27ba4617b06b4b5398 - name: cloud-versioning-demo url: s3://cloud-versioning-demo type: url meta: isdir: true size: 323919 nfiles: 33 files: - relpath: myproject/model.pt meta: size: 106433 version_id: 5qrtnhnQ4fBzV73kqqK6pMGhTOzd_IPr etag: 3bc0028677ce6fb65bec8090c248b002 # truncated ``` The pipeline stages keep them in `dataset` section of individual deps. ```console dvc stage add -n train -d ds://dogs python train.py cat dvc.yaml ``` ```yaml # truncated stages: train: cmd: python train.py deps: - ds://dogs ``` When it is reproduced, the `dvc.lock` will look something like follows: ```yaml # truncated stages: train: cmd: python train.py deps: - path: ds://dogs dataset: name: dogs url: dvcx://dogs type: dvcx version: 3 created_at: '2023-12-11T10:32:05.942708+00:00' ``` * remodeling * remove non-cloud-versioned remotes * improve deserializing; handle invalidation
145 lines
4.6 KiB
Python
145 lines
4.6 KiB
Python
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
import voluptuous as vol
|
|
|
|
from dvc import dependency, output
|
|
from dvc.annotations import ANNOTATION_SCHEMA, ARTIFACT_SCHEMA
|
|
from dvc.output import (
|
|
CHECKSUMS_SCHEMA,
|
|
CLOUD_SCHEMA,
|
|
DIR_FILES_SCHEMA,
|
|
META_SCHEMA,
|
|
Output,
|
|
)
|
|
from dvc.parsing import DO_KWD, FOREACH_KWD, MATRIX_KWD, VARS_KWD
|
|
from dvc.stage.params import StageParams
|
|
|
|
STAGES = "stages"
|
|
SINGLE_STAGE_SCHEMA = {
|
|
StageParams.PARAM_MD5: output.CHECKSUM_SCHEMA,
|
|
StageParams.PARAM_WDIR: vol.Any(str, None),
|
|
StageParams.PARAM_DEPS: vol.Any([dependency.SCHEMA], None),
|
|
StageParams.PARAM_OUTS: vol.Any([output.SCHEMA], None),
|
|
StageParams.PARAM_LOCKED: bool, # backward compatibility
|
|
StageParams.PARAM_FROZEN: bool,
|
|
StageParams.PARAM_META: object,
|
|
StageParams.PARAM_ALWAYS_CHANGED: bool,
|
|
StageParams.PARAM_DESC: str,
|
|
}
|
|
|
|
DATA_SCHEMA: dict[Any, Any] = {
|
|
**CHECKSUMS_SCHEMA,
|
|
**META_SCHEMA,
|
|
vol.Required("path"): str,
|
|
Output.PARAM_CLOUD: CLOUD_SCHEMA,
|
|
Output.PARAM_FILES: [DIR_FILES_SCHEMA],
|
|
Output.PARAM_HASH: str,
|
|
**dependency.DatasetDependency.DATASET_SCHEMA,
|
|
}
|
|
LOCK_FILE_STAGE_SCHEMA = {
|
|
vol.Required(StageParams.PARAM_CMD): vol.Any(str, list),
|
|
StageParams.PARAM_DEPS: [DATA_SCHEMA],
|
|
StageParams.PARAM_PARAMS: {str: {str: object}},
|
|
StageParams.PARAM_OUTS: [DATA_SCHEMA],
|
|
}
|
|
|
|
LOCKFILE_STAGES_SCHEMA = {str: LOCK_FILE_STAGE_SCHEMA}
|
|
LOCKFILE_SCHEMA = {
|
|
vol.Required("schema"): vol.Equal("2.0", "invalid schema version"),
|
|
"datasets": object,
|
|
STAGES: LOCKFILE_STAGES_SCHEMA,
|
|
}
|
|
|
|
OUT_PSTAGE_DETAILED_SCHEMA = {
|
|
str: {
|
|
**ANNOTATION_SCHEMA, # type: ignore[arg-type]
|
|
Output.PARAM_CACHE: bool,
|
|
Output.PARAM_PERSIST: bool,
|
|
"checkpoint": bool,
|
|
Output.PARAM_REMOTE: str,
|
|
Output.PARAM_PUSH: bool,
|
|
}
|
|
}
|
|
|
|
PLOTS = "plots"
|
|
PLOT_PROPS = {
|
|
Output.PARAM_PLOT_TEMPLATE: str,
|
|
Output.PARAM_PLOT_X: str,
|
|
Output.PARAM_PLOT_Y: str,
|
|
Output.PARAM_PLOT_X_LABEL: str,
|
|
Output.PARAM_PLOT_Y_LABEL: str,
|
|
Output.PARAM_PLOT_TITLE: str,
|
|
Output.PARAM_PLOT_HEADER: bool,
|
|
}
|
|
PLOT_PROPS_SCHEMA = OUT_PSTAGE_DETAILED_SCHEMA[str] | PLOT_PROPS
|
|
PLOT_PSTAGE_SCHEMA = {str: vol.Any(PLOT_PROPS_SCHEMA, [PLOT_PROPS_SCHEMA])}
|
|
|
|
PARAM_PSTAGE_NON_DEFAULT_SCHEMA = {str: [str]}
|
|
|
|
VARS_SCHEMA = [str, dict]
|
|
|
|
STAGE_DEFINITION = {
|
|
MATRIX_KWD: {str: vol.Any(str, list)},
|
|
vol.Required(StageParams.PARAM_CMD): vol.Any(str, list),
|
|
vol.Optional(StageParams.PARAM_WDIR): str,
|
|
vol.Optional(StageParams.PARAM_DEPS): [str],
|
|
vol.Optional(StageParams.PARAM_PARAMS): [vol.Any(str, dict)],
|
|
vol.Optional(VARS_KWD): VARS_SCHEMA,
|
|
vol.Optional(StageParams.PARAM_FROZEN): bool,
|
|
vol.Optional(StageParams.PARAM_META): object,
|
|
vol.Optional(StageParams.PARAM_DESC): str,
|
|
vol.Optional(StageParams.PARAM_ALWAYS_CHANGED): bool,
|
|
vol.Optional(StageParams.PARAM_OUTS): [vol.Any(str, OUT_PSTAGE_DETAILED_SCHEMA)],
|
|
vol.Optional(StageParams.PARAM_METRICS): [vol.Any(str, OUT_PSTAGE_DETAILED_SCHEMA)],
|
|
vol.Optional(StageParams.PARAM_PLOTS): [vol.Any(str, PLOT_PSTAGE_SCHEMA)],
|
|
}
|
|
|
|
|
|
def either_or(primary, fallback, fallback_includes=None):
|
|
def validator(data):
|
|
schema = primary
|
|
if isinstance(data, Mapping) and set(fallback_includes or []) & data.keys():
|
|
schema = fallback
|
|
return vol.Schema(schema)(data)
|
|
|
|
return validator
|
|
|
|
|
|
PLOT_DEFINITION = {
|
|
Output.PARAM_PLOT_X: vol.Any(str, {str: str}),
|
|
Output.PARAM_PLOT_Y: vol.Any(str, [str], {str: vol.Any(str, [str])}),
|
|
Output.PARAM_PLOT_X_LABEL: str,
|
|
Output.PARAM_PLOT_Y_LABEL: str,
|
|
Output.PARAM_PLOT_TITLE: str,
|
|
Output.PARAM_PLOT_TEMPLATE: str,
|
|
}
|
|
SINGLE_PLOT_SCHEMA = {vol.Required(str): vol.Any(PLOT_DEFINITION, None)}
|
|
ARTIFACTS = "artifacts"
|
|
SINGLE_ARTIFACT_SCHEMA = vol.Schema({str: ARTIFACT_SCHEMA})
|
|
FOREACH_IN = {
|
|
vol.Required(FOREACH_KWD): vol.Any(dict, list, str),
|
|
vol.Required(DO_KWD): STAGE_DEFINITION,
|
|
}
|
|
SINGLE_PIPELINE_STAGE_SCHEMA = {
|
|
str: either_or(STAGE_DEFINITION, FOREACH_IN, [FOREACH_KWD, DO_KWD])
|
|
}
|
|
|
|
DATASET_SCHEMA = vol.Schema(
|
|
{vol.Required("type"): str, vol.Required("name"): str}, extra=vol.ALLOW_EXTRA
|
|
)
|
|
MULTI_STAGE_SCHEMA = {
|
|
"datasets": object,
|
|
PLOTS: [vol.Any(str, SINGLE_PLOT_SCHEMA)],
|
|
STAGES: SINGLE_PIPELINE_STAGE_SCHEMA,
|
|
VARS_KWD: VARS_SCHEMA,
|
|
StageParams.PARAM_PARAMS: [str],
|
|
StageParams.PARAM_METRICS: [str],
|
|
ARTIFACTS: SINGLE_ARTIFACT_SCHEMA,
|
|
}
|
|
|
|
COMPILED_SINGLE_STAGE_SCHEMA = vol.Schema(SINGLE_STAGE_SCHEMA)
|
|
COMPILED_MULTI_STAGE_SCHEMA = vol.Schema(MULTI_STAGE_SCHEMA)
|
|
COMPILED_LOCK_FILE_STAGE_SCHEMA = vol.Schema(LOCK_FILE_STAGE_SCHEMA)
|
|
COMPILED_LOCKFILE_SCHEMA = vol.Schema(LOCKFILE_SCHEMA)
|