mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-14 07:18:27 +01:00
Saugat Pachhai
99b58d3d05
* refactor multistage load for params and outputs * tests: load params * tests: output loading from pipeline file * fix test * fix typo in name * split params load * rename params func s/inject_values/fill_values * fix tests * simplify loads_params and output.load_from_pipeline * address @pared's suggestions for tests
116 lines
3.5 KiB
Python
116 lines
3.5 KiB
Python
from collections import defaultdict
|
|
from urllib.parse import urlparse
|
|
|
|
import dvc.output as output
|
|
from dvc.dependency.azure import AzureDependency
|
|
from dvc.dependency.gs import GSDependency
|
|
from dvc.dependency.hdfs import HDFSDependency
|
|
from dvc.dependency.http import HTTPDependency
|
|
from dvc.dependency.https import HTTPSDependency
|
|
from dvc.dependency.local import LocalDependency
|
|
from dvc.dependency.param import ParamsDependency
|
|
from dvc.dependency.s3 import S3Dependency
|
|
from dvc.dependency.ssh import SSHDependency
|
|
from dvc.output.base import BaseOutput
|
|
from dvc.remote import Remote
|
|
from dvc.scheme import Schemes
|
|
|
|
from .repo import RepoDependency
|
|
|
|
DEPS = [
|
|
AzureDependency,
|
|
GSDependency,
|
|
HDFSDependency,
|
|
HTTPDependency,
|
|
HTTPSDependency,
|
|
S3Dependency,
|
|
SSHDependency,
|
|
# NOTE: LocalDependency is the default choice
|
|
]
|
|
|
|
DEP_MAP = {
|
|
Schemes.LOCAL: LocalDependency,
|
|
Schemes.SSH: SSHDependency,
|
|
Schemes.S3: S3Dependency,
|
|
Schemes.AZURE: AzureDependency,
|
|
Schemes.GS: GSDependency,
|
|
Schemes.HDFS: HDFSDependency,
|
|
Schemes.HTTP: HTTPDependency,
|
|
Schemes.HTTPS: HTTPSDependency,
|
|
}
|
|
|
|
|
|
# NOTE: schema for dependencies is basically the same as for outputs, but
|
|
# without output-specific entries like 'cache' (whether or not output is
|
|
# cached, see -o and -O flags for `dvc run`) and 'metric' (whether or not
|
|
# output is a metric file and how to parse it, see `-M` flag for `dvc run`).
|
|
SCHEMA = output.SCHEMA.copy()
|
|
del SCHEMA[BaseOutput.PARAM_CACHE]
|
|
del SCHEMA[BaseOutput.PARAM_METRIC]
|
|
SCHEMA.update(RepoDependency.REPO_SCHEMA)
|
|
SCHEMA.update(ParamsDependency.PARAM_SCHEMA)
|
|
|
|
|
|
def _get(stage, p, info):
|
|
parsed = urlparse(p) if p else None
|
|
if parsed and parsed.scheme == "remote":
|
|
remote = Remote(stage.repo, name=parsed.netloc)
|
|
return DEP_MAP[remote.scheme](stage, p, info, remote=remote)
|
|
|
|
if info and info.get(RepoDependency.PARAM_REPO):
|
|
repo = info.pop(RepoDependency.PARAM_REPO)
|
|
return RepoDependency(repo, stage, p, info)
|
|
|
|
if info and info.get(ParamsDependency.PARAM_PARAMS):
|
|
params = info.pop(ParamsDependency.PARAM_PARAMS)
|
|
return ParamsDependency(stage, p, params)
|
|
|
|
for d in DEPS:
|
|
if d.supported(p):
|
|
return d(stage, p, info)
|
|
return LocalDependency(stage, p, info)
|
|
|
|
|
|
def loadd_from(stage, d_list):
|
|
ret = []
|
|
for d in d_list:
|
|
p = d.pop(BaseOutput.PARAM_PATH, None)
|
|
ret.append(_get(stage, p, d))
|
|
return ret
|
|
|
|
|
|
def loads_from(stage, s_list, erepo=None):
|
|
ret = []
|
|
for s in s_list:
|
|
info = {RepoDependency.PARAM_REPO: erepo} if erepo else {}
|
|
ret.append(_get(stage, s, info))
|
|
return ret
|
|
|
|
|
|
def _merge_params(s_list):
|
|
d = defaultdict(list)
|
|
default_file = ParamsDependency.DEFAULT_PARAMS_FILE
|
|
for key in s_list:
|
|
if isinstance(key, str):
|
|
d[default_file].append(key)
|
|
continue
|
|
if not isinstance(key, dict):
|
|
msg = "Only list of str/dict is supported. Got: "
|
|
msg += f"'{type(key).__name__}'."
|
|
raise ValueError(msg)
|
|
|
|
for k, params in key.items():
|
|
if not isinstance(params, list):
|
|
msg = "Expected list of params for custom params file "
|
|
msg += f"'{k}', got '{type(params).__name__}'."
|
|
raise ValueError(msg)
|
|
d[k].extend(params)
|
|
return d
|
|
|
|
|
|
def loads_params(stage, s_list):
|
|
d = _merge_params(s_list)
|
|
return [
|
|
ParamsDependency(stage, path, params) for path, params in d.items()
|
|
]
|