Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-09 23:10:24 +01:00
Code Issues Packages Projects Releases Wiki Activity

fs: deps/outs: get rid of types (#6054)

Browse Source 
Pre-requisite for removing fs.path_info and switching to regular strings
as paths in filesystems.

With new filesystems we no longer need the zoo of different dependency
and output types, as most of them are now just handled by the underlying
filesystem. Though deps/outs are still due for a refactor in other
aspects.
This commit is contained in:
Ruslan Kuprieiev
2021-05-25 19:25:09 +03:00
committed by GitHub
parent 50ec64b841
commit 8a7ccab82f
63 changed files with 466 additions and 807 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/command/stage.py
+3 -3
View File
@@ -9,7 +9,7 @@ from dvc.utils.cli_parse import parse_params
from dvc.utils.humanize import truncate_text
if TYPE_CHECKING:
from dvc.output.base import BaseOutput
from dvc.output import Output
from dvc.stage import Stage
logger = logging.getLogger(__name__)
@@ -18,7 +18,7 @@ MAX_TEXT_LENGTH = 80
def generate_description(stage: "Stage") -> str:
def part_desc(outs: Iterable["BaseOutput"]) -> str:
def part_desc(outs: Iterable["Output"]) -> str:
return ", ".join(out.def_path for out in outs)
if not stage.deps and not stage.outs:
@@ -27,7 +27,7 @@ def generate_description(stage: "Stage") -> str:
if not stage.outs and stage.deps:
return "Depends on " + part_desc(stage.deps)
def is_plot_or_metric(out: "BaseOutput"):
def is_plot_or_metric(out: "Output"):
return bool(out.plot) or bool(out.metric)
desc: List[str] = []
dvc/dependency/__init__.py
+8 -43
View File
@@ -1,59 +1,25 @@
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.dependency.webdav import WebDAVDependency
from dvc.dependency.webdavs import WebDAVSDependency
from dvc.dependency.webhdfs import WebHDFSDependency
from dvc.output.base import BaseOutput
from dvc.scheme import Schemes
from dvc.output import Output
from ..fs import get_cloud_fs
from .base import Dependency
from .param import ParamsDependency
from .repo import RepoDependency
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,
Schemes.WEBDAV: WebDAVDependency,
Schemes.WEBDAVS: WebDAVSDependency,
Schemes.WEBHDFS: WebHDFSDependency,
}
# 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 metrics 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]
del SCHEMA[BaseOutput.PARAM_DESC]
del SCHEMA[Output.PARAM_CACHE]
del SCHEMA[Output.PARAM_METRIC]
del SCHEMA[Output.PARAM_DESC]
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":
cls, config = get_cloud_fs(stage.repo, name=parsed.netloc)
fs = cls(**config)
return DEP_MAP[fs.scheme](stage, p, info, fs=fs)
if info and info.get(RepoDependency.PARAM_REPO):
repo = info.pop(RepoDependency.PARAM_REPO)
return RepoDependency(repo, stage, p, info)
@@ -62,14 +28,13 @@ def _get(stage, p, info):
params = info.pop(ParamsDependency.PARAM_PARAMS)
return ParamsDependency(stage, p, params)
dep_cls = DEP_MAP.get(parsed.scheme, LocalDependency)
return dep_cls(stage, p, info)
return Dependency(stage, p, info)
def loadd_from(stage, d_list):
ret = []
for d in d_list:
p = d.pop(BaseOutput.PARAM_PATH, None)
p = d.pop(Output.PARAM_PATH, None)
ret.append(_get(stage, p, d))
return ret
dvc/dependency/azure.py
-8
View File
@@ -1,8 +0,0 @@
from dvc.dependency.base import BaseDependency
from dvc.output.base import BaseOutput
from ..fs.azure import AzureFileSystem
class AzureDependency(BaseDependency, BaseOutput):
FS_CLS = AzureFileSystem
dvc/dependency/base.py
+2 -1
View File
@@ -1,6 +1,7 @@
from typing import Type
from dvc.exceptions import DvcException
from dvc.output import Output
class DependencyDoesNotExistError(DvcException):
@@ -20,7 +21,7 @@ class DependencyIsStageFileError(DvcException):
super().__init__(f"DVC file '{path}' cannot be a dependency.")
class BaseDependency:
class Dependency(Output):
IS_DEPENDENCY = True
DoesNotExistError = DependencyDoesNotExistError # type: Type[DvcException]
dvc/dependency/gs.py
-8
View File
@@ -1,8 +0,0 @@
from dvc.dependency.base import BaseDependency
from dvc.output.base import BaseOutput
from ..fs.gs import GSFileSystem
class GSDependency(BaseDependency, BaseOutput):
FS_CLS = GSFileSystem
dvc/dependency/hdfs.py
-6
View File
@@ -1,6 +0,0 @@
from dvc.dependency.base import BaseDependency
from dvc.output.hdfs import HDFSOutput
class HDFSDependency(BaseDependency, HDFSOutput):
pass
dvc/dependency/http.py
-8
View File
@@ -1,8 +0,0 @@
from dvc.dependency.base import BaseDependency
from dvc.output.base import BaseOutput
from ..fs.http import HTTPFileSystem
class HTTPDependency(BaseDependency, BaseOutput):
FS_CLS = HTTPFileSystem
dvc/dependency/https.py
-6
View File
@@ -1,6 +0,0 @@
from ..fs.https import HTTPSFileSystem
from .http import HTTPDependency
class HTTPSDependency(HTTPDependency):
FS_CLS = HTTPSFileSystem
dvc/dependency/local.py
-6
View File
@@ -1,6 +0,0 @@
from dvc.dependency.base import BaseDependency
from dvc.output.local import LocalOutput
class LocalDependency(BaseDependency, LocalOutput):
pass
dvc/dependency/param.py
+3 -3
View File
@@ -4,11 +4,12 @@ from collections import defaultdict
import dpath.util
from voluptuous import Any
from dvc.dependency.local import LocalDependency
from dvc.exceptions import DvcException
from dvc.hash_info import HashInfo
from dvc.utils.serialize import LOADERS, ParseError
from .base import Dependency
class MissingParamsError(DvcException):
pass
@@ -18,7 +19,7 @@ class BadParamFileError(DvcException):
pass
class ParamsDependency(LocalDependency):
class ParamsDependency(Dependency):
PARAM_PARAMS = "params"
PARAM_SCHEMA = {PARAM_PARAMS: Any(dict, list, None)}
DEFAULT_PARAMS_FILE = "params.yaml"
@@ -39,7 +40,6 @@ class ParamsDependency(LocalDependency):
path
or os.path.join(stage.repo.root_dir, self.DEFAULT_PARAMS_FILE),
info=info,
fs=stage.repo.fs,
)
def dumpd(self):
dvc/dependency/repo.py
+3 -3
View File
@@ -2,10 +2,10 @@ from voluptuous import Required
from dvc.path_info import PathInfo
from .local import LocalDependency
from .base import Dependency
class RepoDependency(LocalDependency):
class RepoDependency(Dependency):
PARAM_REPO = "repo"
PARAM_URL = "url"
PARAM_REV = "rev"
@@ -23,7 +23,7 @@ class RepoDependency(LocalDependency):
self.def_repo = def_repo
super().__init__(stage, *args, **kwargs)
def _parse_path(self, fs, path):
def _parse_path(self, fs, path_info):
return None
@property
dvc/dependency/s3.py
-6
View File
@@ -1,6 +0,0 @@
from dvc.dependency.base import BaseDependency
from dvc.output.s3 import S3Output
class S3Dependency(BaseDependency, S3Output):
pass
dvc/dependency/ssh.py
-6
View File
@@ -1,6 +0,0 @@
from dvc.dependency.base import BaseDependency
from dvc.output.ssh import SSHOutput
class SSHDependency(BaseDependency, SSHOutput):
pass
dvc/dependency/webdav.py
-8
View File
@@ -1,8 +0,0 @@
from dvc.dependency.base import BaseDependency
from dvc.output.base import BaseOutput
from ..fs.webdav import WebDAVFileSystem
class WebDAVDependency(BaseDependency, BaseOutput):
FS_CLS = WebDAVFileSystem
dvc/dependency/webdavs.py
-6
View File
@@ -1,6 +0,0 @@
from ..fs.webdavs import WebDAVSFileSystem
from .webdav import WebDAVDependency
class WebDAVSDependency(WebDAVDependency):
FS_CLS = WebDAVSFileSystem
dvc/dependency/webhdfs.py
-6
View File
@@ -1,6 +0,0 @@
from dvc.dependency.base import BaseDependency
from dvc.output.webhdfs import WebHDFSOutput
class WebHDFSDependency(BaseDependency, WebHDFSOutput):
pass
dvc/fs/__init__.py
+7 -4
View File
@@ -82,25 +82,28 @@ def get_cloud_fs(repo, **kwargs):
from dvc.config import ConfigError
from dvc.config_schema import SCHEMA, Invalid
remote_conf = get_fs_config(repo.config, **kwargs)
repo_config = repo.config if repo else {}
core_config = repo_config.get("core", {})
remote_conf = get_fs_config(repo_config, **kwargs)
try:
remote_conf = SCHEMA["remote"][str](remote_conf)
except Invalid as exc:
raise ConfigError(str(exc)) from None
if "jobs" not in remote_conf:
jobs = repo.config["core"].get("jobs")
jobs = core_config.get("jobs")
if jobs:
remote_conf["jobs"] = jobs
if "checksum_jobs" not in remote_conf:
checksum_jobs = repo.config["core"].get("checksum_jobs")
checksum_jobs = core_config.get("checksum_jobs")
if checksum_jobs:
remote_conf["checksum_jobs"] = checksum_jobs
cls = get_fs_cls(remote_conf)
if cls == GDriveFileSystem:
if cls == GDriveFileSystem and repo:
remote_conf["gdrive_credentials_tmp_dir"] = repo.tmp_dir
return cls, remote_conf
dvc/fs/_metadata.py
+2 -2
View File
@@ -1,7 +1,7 @@
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, List
from dvc.output import BaseOutput
from dvc.output import Output
from dvc.path_info import PathInfo
if TYPE_CHECKING:
@@ -33,7 +33,7 @@ class Metadata:
# optional fields
isdir: bool = False # is it a directory?
is_exec: bool = False # is it an executable?
outs: List[BaseOutput] = field(default_factory=list) # list of outputs
outs: List[Output] = field(default_factory=list) # list of outputs
def __post_init__(self):
self.output_exists = bool(self.outs)
dvc/fs/dvc.py
+2 -2
View File
@@ -10,7 +10,7 @@ from ._metadata import Metadata
from .base import BaseFileSystem
if typing.TYPE_CHECKING:
from dvc.output.base import BaseOutput
from dvc.output import Output
logger = logging.getLogger(__name__)
@@ -43,7 +43,7 @@ class DvcFileSystem(BaseFileSystem): # pylint:disable=abstract-method
return outs
def _get_granular_hash(
self, path_info: PathInfo, out: "BaseOutput", remote=None
self, path_info: PathInfo, out: "Output", remote=None
):
assert isinstance(path_info, PathInfo)
# NOTE: use string paths here for performance reasons
dvc/output/base.py → dvc/output.py
+278 -29
View File
@@ -1,10 +1,12 @@
import logging
import os
from collections import defaultdict
from copy import copy
from typing import Type
from urllib.parse import urlparse
from voluptuous import Any
from funcy import collecting, project
from voluptuous import And, Any, Coerce, Length, Lower, Required, SetTo
import dvc.objects as objects
import dvc.prompt as prompt
@@ -16,18 +18,179 @@ from dvc.exceptions import (
MergeError,
RemoteCacheRequiredError,
)
from dvc.hash_info import HashInfo
from dvc.objects import save as osave
from dvc.objects.db import NamedCache
from dvc.objects.errors import ObjectFormatError
from dvc.objects.stage import stage as ostage
from dvc.scheme import Schemes
from ..fs.base import BaseFileSystem
from .fs import get_cloud_fs
from .fs.hdfs import HDFSFileSystem
from .fs.local import LocalFileSystem
from .fs.s3 import S3FileSystem
from .hash_info import HashInfo
from .istextfile import istextfile
from .objects import save as osave
from .objects.db import NamedCache
from .objects.errors import ObjectFormatError
from .objects.stage import stage as ostage
from .scheme import Schemes
from .utils import relpath
from .utils.fs import path_isin
logger = logging.getLogger(__name__)
CHECKSUM_SCHEMA = Any(
None,
And(str, Length(max=0), SetTo(None)),
And(Any(str, And(int, Coerce(str))), Length(min=3), Lower),
)
# NOTE: currently there are only 3 possible checksum names:
#
# 1) md5 (LOCAL, SSH);
# 2) etag (S3);
# 3) checksum (HDFS);
#
# so when a few types of outputs share the same name, we only need
# specify it once.
CHECKSUMS_SCHEMA = {
LocalFileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
S3FileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
HDFSFileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
}
def _get(stage, path, **kwargs):
return Output(stage, path, **kwargs)
def loadd_from(stage, d_list):
ret = []
for d in d_list:
p = d.pop(Output.PARAM_PATH)
cache = d.pop(Output.PARAM_CACHE, True)
metric = d.pop(Output.PARAM_METRIC, False)
plot = d.pop(Output.PARAM_PLOT, False)
persist = d.pop(Output.PARAM_PERSIST, False)
checkpoint = d.pop(Output.PARAM_CHECKPOINT, False)
desc = d.pop(Output.PARAM_DESC, False)
isexec = d.pop(Output.PARAM_ISEXEC, False)
live = d.pop(Output.PARAM_LIVE, False)
ret.append(
_get(
stage,
p,
info=d,
cache=cache,
metric=metric,
plot=plot,
persist=persist,
checkpoint=checkpoint,
desc=desc,
isexec=isexec,
live=live,
)
)
return ret
def loads_from(
stage,
s_list,
use_cache=True,
metric=False,
plot=False,
persist=False,
checkpoint=False,
isexec=False,
live=False,
):
return [
_get(
stage,
s,
info={},
cache=use_cache,
metric=metric,
plot=plot,
persist=persist,
checkpoint=checkpoint,
isexec=isexec,
live=live,
)
for s in s_list
]
def _split_dict(d, keys):
return project(d, keys), project(d, d.keys() - keys)
def _merge_data(s_list):
d = defaultdict(dict)
for key in s_list:
if isinstance(key, str):
d[key].update({})
continue
if not isinstance(key, dict):
raise ValueError(f"'{type(key).__name__}' not supported.")
for k, flags in key.items():
if not isinstance(flags, dict):
raise ValueError(
f"Expected dict for '{k}', got: '{type(flags).__name__}'"
)
d[k].update(flags)
return d
@collecting
def load_from_pipeline(stage, data, typ="outs"):
if typ not in (
stage.PARAM_OUTS,
stage.PARAM_METRICS,
stage.PARAM_PLOTS,
stage.PARAM_LIVE,
):
raise ValueError(f"'{typ}' key is not allowed for pipeline files.")
metric = typ == stage.PARAM_METRICS
plot = typ == stage.PARAM_PLOTS
live = typ == stage.PARAM_LIVE
if live:
# `live` is single object
data = [data]
d = _merge_data(data)
for path, flags in d.items():
plt_d, live_d = {}, {}
if plot:
from dvc.schema import PLOT_PROPS
plt_d, flags = _split_dict(flags, keys=PLOT_PROPS.keys())
if live:
from dvc.schema import LIVE_PROPS
live_d, flags = _split_dict(flags, keys=LIVE_PROPS.keys())
extra = project(
flags,
[
Output.PARAM_CACHE,
Output.PARAM_PERSIST,
Output.PARAM_CHECKPOINT,
],
)
yield _get(
stage,
path,
info={},
plot=plt_d or plot,
metric=metric,
live=live_d or live,
**extra,
)
class OutputDoesNotExistError(DvcException):
def __init__(self, path):
msg = f"output '{path}' does not exist"
@@ -61,11 +224,9 @@ class OutputIsIgnoredError(DvcException):
super().__init__(f"Path '{match.file}' is ignored by\n{lines}")
class BaseOutput:
class Output:
IS_DEPENDENCY = False
FS_CLS = BaseFileSystem
PARAM_PATH = "path"
PARAM_CACHE = "cache"
PARAM_CHECKPOINT = "checkpoint"
@@ -108,7 +269,6 @@ class BaseOutput:
stage,
path,
info=None,
fs=None,
cache=True,
metric=False,
plot=False,
@@ -119,10 +279,20 @@ class BaseOutput:
isexec=False,
):
self.repo = stage.repo if stage else None
if fs:
self.fs = fs
fs_cls, fs_config = get_cloud_fs(self.repo, url=path)
self.fs = fs_cls(**fs_config)
path_info = self.fs.path_info
if (
self.fs.scheme == "local"
and stage
and path_isin(path, stage.repo.root_dir)
):
self.def_path = relpath(path, stage.wdir)
else:
self.fs = self.FS_CLS()
self.def_path = path
self._validate_output_path(path, stage)
# This output (and dependency) objects have too many paths/urls
# here is a list and comments:
@@ -135,7 +305,6 @@ class BaseOutput:
# By resolved path, which contains actual location,
# should be absolute and don't contain remote:// refs.
self.stage = stage
self.def_path = path
self.hash_info = HashInfo.from_dict(info)
self.use_cache = False if self.IS_DEPENDENCY else cache
self.metric = False if self.IS_DEPENDENCY else metric
@@ -145,18 +314,30 @@ class BaseOutput:
self.live = live
self.desc = desc
self.path_info = self._parse_path(fs, path)
self.path_info = self._parse_path(self.fs, path_info)
if self.use_cache and self.odb is None:
raise RemoteCacheRequiredError(self.path_info)
self.obj = None
self.isexec = False if self.IS_DEPENDENCY else isexec
def _parse_path(self, fs, path):
if fs:
parsed = urlparse(path)
return fs.path_info / parsed.path.lstrip("/")
return self.FS_CLS.PATH_CLS(path)
def _parse_path(self, fs, path_info):
if fs.scheme != "local":
return path_info
parsed = urlparse(self.def_path)
if parsed.scheme != "remote":
# NOTE: we can path either from command line or .dvc file,
# so we should expect both posix and windows style paths.
# PathInfo accepts both, i.e. / works everywhere, \ only on win.
#
# FIXME: if we have Windows path containing / or posix one with \
# then we have #2059 bug and can't really handle that.
if self.stage and not path_info.is_absolute():
path_info = self.stage.wdir / path_info
abs_p = os.path.abspath(os.path.normpath(path_info))
return fs.PATH_CLS(abs_p)
def __repr__(self):
return "{class_name}: '{def_path}'".format(
@@ -164,15 +345,40 @@ class BaseOutput:
)
def __str__(self):
return self.def_path
if self.fs.scheme != "local":
return self.def_path
if (
not self.repo
or urlparse(self.def_path).scheme == "remote"
or os.path.isabs(self.def_path)
):
return str(self.def_path)
cur_dir = os.getcwd()
if path_isin(cur_dir, self.repo.root_dir):
return relpath(self.path_info, cur_dir)
return relpath(self.path_info, self.repo.root_dir)
@property
def scheme(self):
return self.FS_CLS.scheme
return self.fs.scheme
@property
def is_in_repo(self):
return False
if self.fs.scheme != "local":
return False
if urlparse(self.def_path).scheme == "remote":
return False
if os.path.isabs(self.def_path):
return False
return self.repo and path_isin(
os.path.realpath(self.path_info), self.repo.root_dir
)
@property
def use_scm_ignore(self):
@@ -265,6 +471,8 @@ class BaseOutput:
@property
def dvcignore(self):
if self.fs.scheme == "local":
return self.repo.dvcignore
return None
@property
@@ -371,7 +579,13 @@ class BaseOutput:
def dumpd(self):
ret = copy(self.hash_info.to_dict())
ret[self.PARAM_PATH] = self.def_path
if self.is_in_repo:
path = self.path_info.relpath(self.stage.wdir).as_posix()
else:
path = self.def_path
ret[self.PARAM_PATH] = path
if self.IS_DEPENDENCY:
return ret
@@ -410,7 +624,27 @@ class BaseOutput:
return ret
def verify_metric(self):
raise DvcException(f"verify metric is not supported for {self.scheme}")
if self.fs.scheme != "local":
raise DvcException(
f"verify metric is not supported for {self.scheme}"
)
if not self.metric or self.plot:
return
path = os.fspath(self.path_info)
if not os.path.exists(path):
return
name = "metrics" if self.metric else "plot"
if os.path.isdir(path):
msg = "directory '%s' cannot be used as %s."
logger.debug(msg, str(self.path_info), name)
return
if not istextfile(path, self.fs):
msg = "binary file '{}' cannot be used as {}."
raise DvcException(msg.format(self.path_info, name))
def download(self, to, jobs=None):
self.fs.download(self.path_info, to.path_info, jobs=jobs)
@@ -505,8 +739,6 @@ class BaseOutput:
def transfer(
self, source, odb=None, jobs=None, update=False, no_progress_bar=False
):
from dvc.fs import get_cloud_fs
if odb is None:
odb = self.odb
@@ -738,3 +970,20 @@ class BaseOutput:
self.hash_info = merge(
self.odb, ancestor_info, self.hash_info, other.hash_info
)
@property
def fspath(self):
return self.path_info.fspath
SCHEMA = CHECKSUMS_SCHEMA.copy()
SCHEMA[Required(Output.PARAM_PATH)] = str
SCHEMA[Output.PARAM_CACHE] = bool
SCHEMA[Output.PARAM_METRIC] = Output.METRIC_SCHEMA
SCHEMA[Output.PARAM_PLOT] = bool
SCHEMA[Output.PARAM_PERSIST] = bool
SCHEMA[Output.PARAM_CHECKPOINT] = bool
SCHEMA[HashInfo.PARAM_SIZE] = int
SCHEMA[HashInfo.PARAM_NFILES] = int
SCHEMA[Output.PARAM_DESC] = str
SCHEMA[Output.PARAM_ISEXEC] = bool
dvc/output/__init__.py
-241
View File
@@ -1,241 +0,0 @@
from collections import defaultdict
from urllib.parse import urlparse
from funcy import collecting, project
from voluptuous import And, Any, Coerce, Length, Lower, Required, SetTo
from dvc.hash_info import HashInfo
from dvc.output.base import BaseOutput
from dvc.output.hdfs import HDFSOutput
from dvc.output.local import LocalOutput
from dvc.output.s3 import S3Output
from dvc.output.ssh import SSHOutput
from dvc.output.webhdfs import WebHDFSOutput
from dvc.scheme import Schemes
from ..fs import get_cloud_fs
from ..fs.hdfs import HDFSFileSystem
from ..fs.local import LocalFileSystem
from ..fs.s3 import S3FileSystem
from ..fs.webhdfs import WebHDFSFileSystem
OUTS_MAP = {
Schemes.HDFS: HDFSOutput,
Schemes.S3: S3Output,
Schemes.SSH: SSHOutput,
Schemes.LOCAL: LocalOutput,
Schemes.WEBHDFS: WebHDFSOutput,
}
CHECKSUM_SCHEMA = Any(
None,
And(str, Length(max=0), SetTo(None)),
And(Any(str, And(int, Coerce(str))), Length(min=3), Lower),
)
# NOTE: currently there are only 3 possible checksum names:
#
# 1) md5 (LOCAL, SSH);
# 2) etag (S3);
# 3) checksum (HDFS);
#
# so when a few types of outputs share the same name, we only need
# specify it once.
CHECKSUMS_SCHEMA = {
LocalFileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
S3FileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
HDFSFileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
WebHDFSFileSystem.PARAM_CHECKSUM: CHECKSUM_SCHEMA,
}
SCHEMA = CHECKSUMS_SCHEMA.copy()
SCHEMA[Required(BaseOutput.PARAM_PATH)] = str
SCHEMA[BaseOutput.PARAM_CACHE] = bool
SCHEMA[BaseOutput.PARAM_METRIC] = BaseOutput.METRIC_SCHEMA
SCHEMA[BaseOutput.PARAM_PLOT] = bool
SCHEMA[BaseOutput.PARAM_PERSIST] = bool
SCHEMA[BaseOutput.PARAM_CHECKPOINT] = bool
SCHEMA[HashInfo.PARAM_SIZE] = int
SCHEMA[HashInfo.PARAM_NFILES] = int
SCHEMA[BaseOutput.PARAM_DESC] = str
SCHEMA[BaseOutput.PARAM_ISEXEC] = bool
def _get(
stage,
p,
info=None,
cache=True,
metric=False,
plot=False,
persist=False,
checkpoint=False,
live=False,
desc=None,
isexec=False,
):
parsed = urlparse(p)
if parsed.scheme == "remote":
cls, config = get_cloud_fs(stage.repo, name=parsed.netloc)
fs = cls(**config)
return OUTS_MAP[fs.scheme](
stage,
p,
info,
cache=cache,
fs=fs,
metric=metric,
plot=plot,
persist=persist,
checkpoint=checkpoint,
live=live,
desc=desc,
isexec=isexec,
)
out_cls = OUTS_MAP.get(parsed.scheme, LocalOutput)
return out_cls(
stage,
p,
info,
cache=cache,
fs=None,
metric=metric,
plot=plot,
persist=persist,
checkpoint=checkpoint,
live=live,
desc=desc,
isexec=isexec,
)
def loadd_from(stage, d_list):
ret = []
for d in d_list:
p = d.pop(BaseOutput.PARAM_PATH)
cache = d.pop(BaseOutput.PARAM_CACHE, True)
metric = d.pop(BaseOutput.PARAM_METRIC, False)
plot = d.pop(BaseOutput.PARAM_PLOT, False)
persist = d.pop(BaseOutput.PARAM_PERSIST, False)
checkpoint = d.pop(BaseOutput.PARAM_CHECKPOINT, False)
desc = d.pop(BaseOutput.PARAM_DESC, False)
isexec = d.pop(BaseOutput.PARAM_ISEXEC, False)
live = d.pop(BaseOutput.PARAM_LIVE, False)
ret.append(
_get(
stage,
p,
info=d,
cache=cache,
metric=metric,
plot=plot,
persist=persist,
checkpoint=checkpoint,
desc=desc,
isexec=isexec,
live=live,
)
)
return ret
def loads_from(
stage,
s_list,
use_cache=True,
metric=False,
plot=False,
persist=False,
checkpoint=False,
isexec=False,
live=False,
):
return [
_get(
stage,
s,
info={},
cache=use_cache,
metric=metric,
plot=plot,
persist=persist,
checkpoint=checkpoint,
isexec=isexec,
live=live,
)
for s in s_list
]
def _split_dict(d, keys):
return project(d, keys), project(d, d.keys() - keys)
def _merge_data(s_list):
d = defaultdict(dict)
for key in s_list:
if isinstance(key, str):
d[key].update({})
continue
if not isinstance(key, dict):
raise ValueError(f"'{type(key).__name__}' not supported.")
for k, flags in key.items():
if not isinstance(flags, dict):
raise ValueError(
f"Expected dict for '{k}', got: '{type(flags).__name__}'"
)
d[k].update(flags)
return d
@collecting
def load_from_pipeline(stage, data, typ="outs"):
if typ not in (
stage.PARAM_OUTS,
stage.PARAM_METRICS,
stage.PARAM_PLOTS,
stage.PARAM_LIVE,
):
raise ValueError(f"'{typ}' key is not allowed for pipeline files.")
metric = typ == stage.PARAM_METRICS
plot = typ == stage.PARAM_PLOTS
live = typ == stage.PARAM_LIVE
if live:
# `live` is single object
data = [data]
d = _merge_data(data)
for path, flags in d.items():
plt_d, live_d = {}, {}
if plot:
from dvc.schema import PLOT_PROPS
plt_d, flags = _split_dict(flags, keys=PLOT_PROPS.keys())
if live:
from dvc.schema import LIVE_PROPS
live_d, flags = _split_dict(flags, keys=LIVE_PROPS.keys())
extra = project(
flags,
[
BaseOutput.PARAM_CACHE,
BaseOutput.PARAM_PERSIST,
BaseOutput.PARAM_CHECKPOINT,
],
)
yield _get(
stage,
path,
{},
plot=plt_d or plot,
metric=metric,
live=live_d or live,
**extra,
)
dvc/output/hdfs.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.output.base import BaseOutput
from ..fs.hdfs import HDFSFileSystem
class HDFSOutput(BaseOutput):
FS_CLS = HDFSFileSystem
dvc/output/local.py
-111
View File
@@ -1,111 +0,0 @@
import logging
import os
from urllib.parse import urlparse
from dvc.exceptions import DvcException
from dvc.istextfile import istextfile
from dvc.output.base import BaseOutput
from dvc.utils import relpath
from dvc.utils.fs import path_isin
from ..fs.local import LocalFileSystem
logger = logging.getLogger(__name__)
class LocalOutput(BaseOutput):
FS_CLS = LocalFileSystem
sep = os.sep
def __init__(self, stage, path, *args, **kwargs):
if stage and path_isin(path, stage.repo.root_dir):
path = relpath(path, stage.wdir)
super().__init__(stage, path, *args, **kwargs)
if (
self.is_in_repo
and self.repo
and isinstance(self.repo.fs, LocalFileSystem)
):
self.fs = self.repo.fs
def _parse_path(self, fs, path):
parsed = urlparse(path)
if parsed.scheme == "remote":
p = fs.path_info / parsed.path.lstrip("/")
else:
# NOTE: we can path either from command line or .dvc file,
# so we should expect both posix and windows style paths.
# PathInfo accepts both, i.e. / works everywhere, \ only on win.
#
# FIXME: if we have Windows path containing / or posix one with \
# then we have #2059 bug and can't really handle that.
p = self.FS_CLS.PATH_CLS(path)
if self.stage and not p.is_absolute():
p = self.stage.wdir / p
abs_p = os.path.abspath(os.path.normpath(p))
return self.FS_CLS.PATH_CLS(abs_p)
def __str__(self):
if (
not self.repo
or urlparse(self.def_path).scheme == "remote"
or os.path.isabs(self.def_path)
):
return str(self.def_path)
cur_dir = os.getcwd()
if path_isin(cur_dir, self.repo.root_dir):
return relpath(self.path_info, cur_dir)
return relpath(self.path_info, self.repo.root_dir)
@property
def fspath(self):
return self.path_info.fspath
@property
def is_in_repo(self):
if urlparse(self.def_path).scheme == "remote":
return False
if os.path.isabs(self.def_path):
return False
return self.repo and path_isin(
os.path.realpath(self.path_info), self.repo.root_dir
)
def dumpd(self):
ret = super().dumpd()
if self.is_in_repo:
path = self.path_info.relpath(self.stage.wdir).as_posix()
else:
path = self.def_path
ret[self.PARAM_PATH] = path
return ret
def verify_metric(self):
if not self.metric or self.plot:
return
path = os.fspath(self.path_info)
if not os.path.exists(path):
return
name = "metrics" if self.metric else "plot"
if os.path.isdir(path):
msg = "directory '%s' cannot be used as %s."
logger.debug(msg, str(self.path_info), name)
return
if not istextfile(path, self.fs):
msg = "binary file '{}' cannot be used as {}."
raise DvcException(msg.format(self.path_info, name))
@property
def dvcignore(self):
return self.repo.dvcignore
dvc/output/s3.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.output.base import BaseOutput
from ..fs.s3 import S3FileSystem
class S3Output(BaseOutput):
FS_CLS = S3FileSystem
dvc/output/ssh.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.output.base import BaseOutput
from ..fs.ssh import SSHFileSystem
class SSHOutput(BaseOutput):
FS_CLS = SSHFileSystem
dvc/output/webhdfs.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.output.base import BaseOutput
from ..fs.webhdfs import WebHDFSFileSystem
class WebHDFSOutput(BaseOutput):
FS_CLS = WebHDFSFileSystem
dvc/repo/add.py
+1 -1
View File
@@ -156,7 +156,7 @@ def _process_stages(
link_failures = []
from dvc.dvcfile import Dvcfile
from ..output.base import OutputDoesNotExistError
from ..output import OutputDoesNotExistError
if to_remote or to_cache:
# Already verified in the add()
dvc/repo/collect.py
+3 -3
View File
@@ -6,14 +6,14 @@ from dvc.path_info import PathInfo
from dvc.types import DvcPath
if TYPE_CHECKING:
from dvc.output.base import BaseOutput
from dvc.output import Output
from dvc.repo import Repo
logger = logging.getLogger(__name__)
FilterFn = Callable[["BaseOutput"], bool]
Outputs = List["BaseOutput"]
FilterFn = Callable[["Output"], bool]
Outputs = List["Output"]
DvcPaths = List[DvcPath]
dvc/repo/live.py
+4 -4
View File
@@ -8,7 +8,7 @@ from dvc.exceptions import MetricDoesNotExistError, MetricsError
logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from dvc.output import BaseOutput
from dvc.output import Output
from dvc.path_info import PathInfo
from dvc.repo import Repo
@@ -24,13 +24,13 @@ def create_summary(out):
logger.info(f"\nfile://{os.path.abspath(html_path)}")
def summary_path_info(out: "BaseOutput") -> Optional["PathInfo"]:
from dvc.output import BaseOutput
def summary_path_info(out: "Output") -> Optional["PathInfo"]:
from dvc.output import Output
assert out.live
has_summary = True
if isinstance(out.live, dict):
has_summary = out.live.get(BaseOutput.PARAM_LIVE_SUMMARY, True)
has_summary = out.live.get(Output.PARAM_LIVE_SUMMARY, True)
if has_summary:
return out.path_info.with_suffix(".json")
return None
dvc/repo/metrics/show.py
+3 -3
View File
@@ -7,7 +7,7 @@ from dvc.exceptions import (
NoMetricsParsedError,
)
from dvc.fs.repo import RepoFileSystem
from dvc.output import BaseOutput
from dvc.output import Output
from dvc.path_info import PathInfo
from dvc.repo import locked
from dvc.repo.collect import collect
@@ -18,11 +18,11 @@ from dvc.utils.serialize import YAMLFileCorruptedError, load_yaml
logger = logging.getLogger(__name__)
def _is_metric(out: BaseOutput) -> bool:
def _is_metric(out: Output) -> bool:
return bool(out.metric) or bool(out.live)
def _to_path_infos(metrics: List[BaseOutput]) -> List[PathInfo]:
def _to_path_infos(metrics: List[Output]) -> List[PathInfo]:
result = []
for out in metrics:
if out.metric:
dvc/repo/params/show.py
+3 -3
View File
@@ -13,7 +13,7 @@ from dvc.stage import PipelineStage
from dvc.utils.serialize import LOADERS, ParseError
if TYPE_CHECKING:
from dvc.output.base import BaseOutput
from dvc.output import Output
from dvc.repo import Repo
from dvc.types import DvcPath
@@ -24,13 +24,13 @@ class NoParamsError(DvcException):
pass
def _is_params(dep: "BaseOutput"):
def _is_params(dep: "Output"):
return isinstance(dep, ParamsDependency)
def _collect_configs(
repo: "Repo", rev, targets=None
) -> Tuple[List["BaseOutput"], List["DvcPath"]]:
) -> Tuple[List["Output"], List["DvcPath"]]:
params, path_infos = collect(
repo,
targets=targets or [],
dvc/repo/plots/__init__.py
+3 -3
View File
@@ -14,7 +14,7 @@ from dvc.types import StrPath
from dvc.utils import relpath
if TYPE_CHECKING:
from dvc.output import BaseOutput
from dvc.output import Output
from dvc.repo import Repo
logger = logging.getLogger(__name__)
@@ -254,7 +254,7 @@ class Plots:
write(path, plots, metrics, html_template_path)
def _is_plot(out: "BaseOutput") -> bool:
def _is_plot(out: "Output") -> bool:
return bool(out.plot) or bool(out.live)
@@ -278,7 +278,7 @@ def _collect_plots(
return result
def _plot_props(out: "BaseOutput") -> Dict:
def _plot_props(out: "Output") -> Dict:
from dvc.schema import PLOT_PROPS
if not (out.plot or out.live):
dvc/schema.py
+15 -15
View File
@@ -2,7 +2,7 @@ from voluptuous import Any, Optional, Required, Schema
from dvc import dependency, output
from dvc.hash_info import HashInfo
from dvc.output import CHECKSUMS_SCHEMA, BaseOutput
from dvc.output import CHECKSUMS_SCHEMA, Output
from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD
from dvc.parsing.versions import SCHEMA_KWD, lockfile_version_schema
from dvc.stage.params import StageParams
@@ -26,7 +26,7 @@ DATA_SCHEMA = {
Required("path"): str,
HashInfo.PARAM_SIZE: int,
HashInfo.PARAM_NFILES: int,
BaseOutput.PARAM_ISEXEC: bool,
Output.PARAM_ISEXEC: bool,
}
LOCK_FILE_STAGE_SCHEMA = {
Required(StageParams.PARAM_CMD): Any(str, list),
@@ -45,28 +45,28 @@ LOCKFILE_V2_SCHEMA = {
OUT_PSTAGE_DETAILED_SCHEMA = {
str: {
BaseOutput.PARAM_CACHE: bool,
BaseOutput.PARAM_PERSIST: bool,
BaseOutput.PARAM_CHECKPOINT: bool,
BaseOutput.PARAM_DESC: str,
Output.PARAM_CACHE: bool,
Output.PARAM_PERSIST: bool,
Output.PARAM_CHECKPOINT: bool,
Output.PARAM_DESC: str,
}
}
PLOT_PROPS = {
BaseOutput.PARAM_PLOT_TEMPLATE: str,
BaseOutput.PARAM_PLOT_X: str,
BaseOutput.PARAM_PLOT_Y: str,
BaseOutput.PARAM_PLOT_X_LABEL: str,
BaseOutput.PARAM_PLOT_Y_LABEL: str,
BaseOutput.PARAM_PLOT_TITLE: str,
BaseOutput.PARAM_PLOT_HEADER: bool,
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: Any(PLOT_PROPS_SCHEMA, [PLOT_PROPS_SCHEMA])}
LIVE_PROPS = {
BaseOutput.PARAM_LIVE_SUMMARY: bool,
BaseOutput.PARAM_LIVE_HTML: bool,
Output.PARAM_LIVE_SUMMARY: bool,
Output.PARAM_LIVE_HTML: bool,
}
LIVE_PROPS_SCHEMA = {**PLOT_PROPS_SCHEMA, **LIVE_PROPS}
LIVE_PSTAGE_SCHEMA = {str: LIVE_PROPS_SCHEMA}
dvc/stage/__init__.py
+5 -5
View File
@@ -268,7 +268,7 @@ class Stage(params.StageParams):
env: Env = {}
if out.live:
from dvc.env import DVCLIVE_HTML, DVCLIVE_PATH, DVCLIVE_SUMMARY
from dvc.output import BaseOutput
from dvc.output import Output
from dvc.schema import LIVE_PROPS
env[DVCLIVE_PATH] = str(out.path_info)
@@ -276,10 +276,10 @@ class Stage(params.StageParams):
config = project(out.live, LIVE_PROPS)
env[DVCLIVE_SUMMARY] = str(
int(config.get(BaseOutput.PARAM_LIVE_SUMMARY, True))
int(config.get(Output.PARAM_LIVE_SUMMARY, True))
)
env[DVCLIVE_HTML] = str(
int(config.get(BaseOutput.PARAM_LIVE_HTML, True))
int(config.get(Output.PARAM_LIVE_HTML, True))
)
elif out.checkpoint and checkpoint_func:
from dvc.env import DVC_CHECKPOINT
@@ -453,7 +453,7 @@ class Stage(params.StageParams):
raise
def save_outs(self, allow_missing=False):
from dvc.output.base import OutputDoesNotExistError
from dvc.output import OutputDoesNotExistError
for out in self.outs:
try:
@@ -484,7 +484,7 @@ class Stage(params.StageParams):
@rwlocked(write=["outs"])
def commit(self, allow_missing=False, filter_info=None):
from dvc.output.base import OutputDoesNotExistError
from dvc.output import OutputDoesNotExistError
link_failures = []
for out in self.filter_outs(filter_info):
dvc/stage/monitor.py
+2 -4
View File
@@ -11,7 +11,7 @@ from dvc.stage.decorators import relock_repo
from dvc.stage.exceptions import StageCmdFailedError
if TYPE_CHECKING:
from dvc.output import BaseOutput
from dvc.output import Output
from dvc.stage import Stage
@@ -84,9 +84,7 @@ class LiveTask(MonitorTask):
SIGNAL_FILE = "DVC_LIVE"
error_cls = LiveKilledError
def __init__(
self, stage: "Stage", out: "BaseOutput", proc: subprocess.Popen
):
def __init__(self, stage: "Stage", out: "Output", proc: subprocess.Popen):
super().__init__(stage, functools.partial(create_summary, out), proc)
def after_run(self):
dvc/stage/serialize.py
+8 -8
View File
@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, List, no_type_check
from funcy import post_processing
from dvc.dependency import ParamsDependency
from dvc.output import BaseOutput
from dvc.output import Output
from dvc.utils.collections import apply_diff
from dvc.utils.serialize import parse_yaml_for_update
@@ -22,12 +22,12 @@ PARAM_PATH = ParamsDependency.PARAM_PATH
PARAM_DEPS = StageParams.PARAM_DEPS
PARAM_OUTS = StageParams.PARAM_OUTS
PARAM_CACHE = BaseOutput.PARAM_CACHE
PARAM_METRIC = BaseOutput.PARAM_METRIC
PARAM_PLOT = BaseOutput.PARAM_PLOT
PARAM_PERSIST = BaseOutput.PARAM_PERSIST
PARAM_CHECKPOINT = BaseOutput.PARAM_CHECKPOINT
PARAM_DESC = BaseOutput.PARAM_DESC
PARAM_CACHE = Output.PARAM_CACHE
PARAM_METRIC = Output.PARAM_METRIC
PARAM_PLOT = Output.PARAM_PLOT
PARAM_PERSIST = Output.PARAM_PERSIST
PARAM_CHECKPOINT = Output.PARAM_CHECKPOINT
PARAM_DESC = Output.PARAM_DESC
DEFAULT_PARAMS_FILE = ParamsDependency.DEFAULT_PARAMS_FILE
@@ -60,7 +60,7 @@ def _serialize_out(out):
@no_type_check
def _serialize_outs(outputs: List[BaseOutput]):
def _serialize_outs(outputs: List[Output]):
outs, metrics, plots, live = [], [], [], None
for out in sort_by_path(outputs):
bucket = outs
dvc/stage/utils.py
+9 -9
View File
@@ -84,7 +84,7 @@ def _load_live_output(
live_html=False,
**kwargs,
):
from dvc.output import BaseOutput, loads_from
from dvc.output import Output, loads_from
outs = []
if live or live_no_cache:
@@ -96,8 +96,8 @@ def _load_live_output(
[path],
use_cache=not bool(live_no_cache),
live={
BaseOutput.PARAM_LIVE_SUMMARY: live_summary,
BaseOutput.PARAM_LIVE_HTML: live_html,
Output.PARAM_LIVE_SUMMARY: live_summary,
Output.PARAM_LIVE_HTML: live_html,
},
)
@@ -170,7 +170,7 @@ def check_missing_outputs(stage):
def compute_md5(stage):
from dvc.output.base import BaseOutput
from dvc.output import Output
from ..utils import dict_md5
@@ -193,11 +193,11 @@ def compute_md5(stage):
exclude=[
stage.PARAM_LOCKED, # backward compatibility
stage.PARAM_FROZEN,
BaseOutput.PARAM_DESC,
BaseOutput.PARAM_METRIC,
BaseOutput.PARAM_PERSIST,
BaseOutput.PARAM_CHECKPOINT,
BaseOutput.PARAM_ISEXEC,
Output.PARAM_DESC,
Output.PARAM_METRIC,
Output.PARAM_PERSIST,
Output.PARAM_CHECKPOINT,
Output.PARAM_ISEXEC,
HashInfo.PARAM_SIZE,
HashInfo.PARAM_NFILES,
],
tests/func/test_add.py
+1 -1
View File
@@ -24,7 +24,7 @@ from dvc.fs.local import LocalFileSystem
from dvc.hash_info import HashInfo
from dvc.main import main
from dvc.objects.db import ODBManager
from dvc.output.base import OutputAlreadyTrackedError, OutputIsStageFileError
from dvc.output import OutputAlreadyTrackedError, OutputIsStageFileError
from dvc.stage import Stage
from dvc.stage.exceptions import (
StageExternalOutputsError,
tests/func/test_commit.py
+1 -1
View File
@@ -4,7 +4,7 @@ import pytest
from dvc.dependency.base import DependencyDoesNotExistError
from dvc.dvcfile import PIPELINE_FILE
from dvc.output.base import OutputDoesNotExistError
from dvc.output import OutputDoesNotExistError
from dvc.stage.exceptions import StageCommitError
from dvc.utils.serialize import dump_yaml, load_yaml
tests/func/test_dependency.py
-21
View File
@@ -1,21 +0,0 @@
import copy
import pytest
from dvc.dependency import DEP_MAP, _get
from dvc.stage import Stage
from tests.func.test_output import TESTS as OUT_TESTS
TESTS = copy.copy(OUT_TESTS)
TESTS.append(("http://example.com/path/to/file", "http"))
TESTS.append(("https://example.com/path/to/file", "https"))
def _get_dep(dvc, path):
return _get(Stage(dvc), path, None)
@pytest.mark.parametrize("url,scheme", TESTS)
def test_scheme(dvc, url, scheme):
# pylint: disable=unidiomatic-typecheck
assert type(_get_dep(dvc, url)) == DEP_MAP[scheme]
tests/func/test_ignore.py
+1 -1
View File
@@ -6,7 +6,7 @@ import pytest
from dvc.exceptions import DvcIgnoreInCollectedDirError
from dvc.ignore import DvcIgnore, DvcIgnorePatterns
from dvc.output.base import OutputIsIgnoredError
from dvc.output import OutputIsIgnoredError
from dvc.pathspec_math import PatternInfo, merge_patterns
from dvc.repo import Repo
from dvc.types import List
tests/func/test_output.py
-30
View File
@@ -1,30 +0,0 @@
import pytest
from dvc.output import OUTS_MAP, _get
from dvc.stage import Stage
MARKERS = [pytest.mark.hdfs]
TESTS = [
("s3://bucket/path", "s3"),
("ssh://example.com:/dir/path", "ssh"),
("path/to/file", "local"),
("path\\to\\file", "local"),
("file", "local"),
("./file", "local"),
(".\\file", "local"),
("../file", "local"),
("..\\file", "local"),
("unknown://path", "local"),
pytest.param("hdfs://example.com/dir/path", "hdfs", marks=MARKERS),
pytest.param("hdfs://example.com/dir/path", "hdfs", marks=MARKERS),
]
def _get_out(dvc, path):
return _get(Stage(dvc), path, None, None, None, None)
@pytest.mark.parametrize("url,scheme", TESTS)
def test_scheme(dvc, url, scheme):
# pylint: disable=unidiomatic-typecheck
assert type(_get_out(dvc, url)) == OUTS_MAP[scheme]
tests/func/test_repro.py
+3 -3
View File
@@ -15,7 +15,7 @@ from dvc.exceptions import (
)
from dvc.fs.local import LocalFileSystem
from dvc.main import main
from dvc.output.base import BaseOutput
from dvc.output import Output
from dvc.stage import Stage
from dvc.stage.exceptions import StageFileDoesNotExistError
from dvc.system import System
@@ -883,9 +883,9 @@ class TestReproAlreadyCached(TestRepro):
)
patch_checkout = patch.object(
BaseOutput,
Output,
"checkout",
side_effect=BaseOutput.checkout,
side_effect=Output.checkout,
autospec=True,
)
tests/func/test_run_cache.py
+2 -2
View File
@@ -150,7 +150,7 @@ def test_memory_runs_of_multiple_stages(tmp_dir, dvc, run_copy, mocker):
def test_restore_pull(tmp_dir, dvc, run_copy, mocker, local_remote):
from dvc.output import base
import dvc.output as dvc_output
tmp_dir.gen("foo", "foo")
stage = run_copy("foo", "bar", name="copy-foo-bar")
@@ -159,7 +159,7 @@ def test_restore_pull(tmp_dir, dvc, run_copy, mocker, local_remote):
mock_restore = mocker.spy(dvc.stage_cache, "restore")
mock_run = mocker.patch("dvc.stage.run.cmd_run")
mock_checkout = mocker.spy(base, "checkout")
mock_checkout = mocker.spy(dvc_output, "checkout")
# removing any information that `dvc` could use to re-generate from
(tmp_dir / "bar").unlink()
tests/func/test_run_single_stage.py
+2 -3
View File
@@ -19,8 +19,7 @@ from dvc.exceptions import (
StagePathAsOutputError,
)
from dvc.main import main
from dvc.output import BaseOutput
from dvc.output.base import OutputIsStageFileError
from dvc.output import Output, OutputIsStageFileError
from dvc.repo import Repo as DvcRepo
from dvc.stage import Stage
from dvc.stage.exceptions import (
@@ -763,7 +762,7 @@ class TestRunPersist(TestDvc):
def stage_should_contain_persist_flag(self, stage_file):
stage_file_content = load_yaml(stage_file)
self.assertEqual(
True, stage_file_content["outs"][0][BaseOutput.PARAM_PERSIST]
True, stage_file_content["outs"][0][Output.PARAM_PERSIST]
)
def should_append_upon_repro(self, file, stage_file):
tests/func/test_stage.py
+6 -6
View File
@@ -6,7 +6,7 @@ import pytest
from dvc.dvcfile import SingleStageFile
from dvc.fs.local import LocalFileSystem
from dvc.main import main
from dvc.output.local import LocalOutput
from dvc.output import Output
from dvc.repo import Repo, lock_repo
from dvc.stage import PipelineStage, Stage
from dvc.stage.exceptions import StageFileFormatError
@@ -55,15 +55,15 @@ def test_empty_list():
def test_list():
lst = [
{LocalOutput.PARAM_PATH: "foo", LocalFileSystem.PARAM_CHECKSUM: "123"},
{LocalOutput.PARAM_PATH: "bar", LocalFileSystem.PARAM_CHECKSUM: None},
{LocalOutput.PARAM_PATH: "baz"},
{Output.PARAM_PATH: "foo", LocalFileSystem.PARAM_CHECKSUM: "123"},
{Output.PARAM_PATH: "bar", LocalFileSystem.PARAM_CHECKSUM: None},
{Output.PARAM_PATH: "baz"},
]
d = {Stage.PARAM_DEPS: lst}
SingleStageFile.validate(d)
lst[0][LocalOutput.PARAM_CACHE] = True
lst[1][LocalOutput.PARAM_CACHE] = False
lst[0][Output.PARAM_CACHE] = True
lst[1][Output.PARAM_CACHE] = False
d = {Stage.PARAM_OUTS: lst}
SingleStageFile.validate(d)
tests/unit/dependency/test_azure.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.dependency.azure import AzureDependency
from tests.unit.dependency.test_local import TestLocalDependency
class TestAzureDependency(TestLocalDependency):
def _get_cls(self):
return AzureDependency
tests/unit/dependency/test_dependency.py
+12
View File
@@ -0,0 +1,12 @@
import pytest
from dvc.dependency import Dependency
from dvc.stage import Stage
def test_save_missing(dvc, mocker):
stage = Stage(dvc)
dep = Dependency(stage, "path")
with mocker.patch.object(dep.fs, "exists", return_value=False):
with pytest.raises(dep.DoesNotExistError):
dep.save()
tests/unit/dependency/test_gs.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.dependency.gs import GSDependency
from tests.unit.dependency.test_local import TestLocalDependency
class TestGSDependency(TestLocalDependency):
def _get_cls(self):
return GSDependency
tests/unit/dependency/test_hdfs.py
-10
View File
@@ -1,10 +0,0 @@
import pytest
from dvc.dependency.hdfs import HDFSDependency
from tests.unit.dependency.test_local import TestLocalDependency
@pytest.mark.hdfs
class TestHDFSDependency(TestLocalDependency):
def _get_cls(self):
return HDFSDependency
tests/unit/dependency/test_http.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.dependency.http import HTTPDependency
from tests.unit.dependency.test_local import TestLocalDependency
class TestHTTPDependency(TestLocalDependency):
def _get_cls(self):
return HTTPDependency
tests/unit/dependency/test_local.py
-20
View File
@@ -1,20 +0,0 @@
import mock
from dvc.dependency import LocalDependency
from dvc.stage import Stage
from tests.basic_env import TestDvc
class TestLocalDependency(TestDvc):
def _get_cls(self):
return LocalDependency
def _get_dependency(self):
stage = Stage(self.dvc)
return self._get_cls()(stage, "path")
def test_save_missing(self):
d = self._get_dependency()
with mock.patch.object(d.fs, "exists", return_value=False):
with self.assertRaises(d.DoesNotExistError):
d.save()
tests/unit/dependency/test_s3.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.dependency.s3 import S3Dependency
from tests.unit.dependency.test_local import TestLocalDependency
class TestS3Dependency(TestLocalDependency):
def _get_cls(self):
return S3Dependency
tests/unit/dependency/test_ssh.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.dependency.ssh import SSHDependency
from tests.unit.dependency.test_local import TestLocalDependency
class TestSSHDependency(TestLocalDependency):
def _get_cls(self):
return SSHDependency
tests/unit/dependency/test_webdav.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.dependency.webdav import WebDAVDependency
from tests.unit.dependency.test_local import TestLocalDependency
class TestWebDAVDependency(TestLocalDependency):
def _get_cls(self):
return WebDAVDependency
tests/unit/fs/test_fs.py
+31
View File
@@ -0,0 +1,31 @@
import pytest
from dvc.fs import get_fs_cls
from dvc.fs.hdfs import HDFSFileSystem
from dvc.fs.http import HTTPFileSystem
from dvc.fs.https import HTTPSFileSystem
from dvc.fs.local import LocalFileSystem
from dvc.fs.s3 import S3FileSystem
from dvc.fs.ssh import SSHFileSystem
@pytest.mark.parametrize(
"url, cls",
[
("s3://bucket/path", S3FileSystem),
("ssh://example.com:/dir/path", SSHFileSystem),
("hdfs://example.com/dir/path", HDFSFileSystem),
("http://example.com/path/to/file", HTTPFileSystem),
("https://example.com/path/to/file", HTTPSFileSystem),
("path/to/file", LocalFileSystem),
("path\\to\\file", LocalFileSystem),
("file", LocalFileSystem),
("./file", LocalFileSystem),
(".\\file", LocalFileSystem),
("../file", LocalFileSystem),
("..\\file", LocalFileSystem),
("unknown://path", LocalFileSystem),
],
)
def test_get_fs_cls(url, cls):
assert get_fs_cls({"url": url}) == cls
tests/unit/output/test_hdfs.py
-10
View File
@@ -1,10 +0,0 @@
import pytest
from dvc.output.hdfs import HDFSOutput
from tests.unit.output.test_local import TestLocalOutput
@pytest.mark.hdfs
class TestHDFSOutput(TestLocalOutput):
def _get_cls(self):
return HDFSOutput
tests/unit/output/test_load.py
+13 -6
View File
@@ -3,7 +3,9 @@
import pytest
from dvc import output
from dvc.output import LocalOutput, S3Output
from dvc.fs.local import LocalFileSystem
from dvc.fs.s3 import S3FileSystem
from dvc.output import Output
from dvc.stage import Stage
@@ -35,7 +37,8 @@ def test_load_from_pipeline(dvc, out_type, type_test_func):
assert len(outs) == 6
for i, out in enumerate(outs, start=1):
assert isinstance(out, LocalOutput)
assert isinstance(out, Output)
assert isinstance(out.fs, LocalFileSystem)
assert out.def_path == f"file{i}"
assert out.use_cache == (out.def_path in cached_outs)
assert out.persist == (out.def_path in persisted_outs)
@@ -55,7 +58,8 @@ def test_load_from_pipeline_accumulates_flag(dvc):
"outs",
)
for out in outs:
assert isinstance(out, LocalOutput)
assert isinstance(out, Output)
assert isinstance(out.fs, LocalFileSystem)
assert not out.plot and not out.metric
assert not out.hash_info
@@ -68,7 +72,8 @@ def test_load_remote_files_from_pipeline(dvc):
(out,) = output.load_from_pipeline(
stage, [{"s3://dvc-test/file.txt": {"cache": False}}], typ="metrics"
)
assert isinstance(out, S3Output)
assert isinstance(out, Output)
assert isinstance(out.fs, S3FileSystem)
assert not out.plot and out.metric
assert not out.persist
assert not out.hash_info
@@ -105,12 +110,14 @@ def test_plots_load_from_pipeline(dvc):
],
"plots",
)
assert isinstance(outs[0], LocalOutput)
assert isinstance(outs[0], Output)
assert isinstance(outs[0].fs, LocalFileSystem)
assert outs[0].use_cache
assert outs[0].plot is True and not outs[0].metric
assert not outs[0].persist
assert isinstance(outs[1], LocalOutput)
assert isinstance(outs[1], Output)
assert isinstance(outs[1].fs, LocalFileSystem)
assert not outs[1].use_cache
assert outs[1].plot == {"x": 3} and not outs[1].metric
assert outs[1].persist
tests/unit/output/test_local.py
+8 -23
View File
@@ -3,42 +3,27 @@ import os
from mock import patch
from dvc.hash_info import HashInfo
from dvc.output import LocalOutput
from dvc.output import Output
from dvc.stage import Stage
from dvc.utils import relpath
from tests.basic_env import TestDvc
class TestLocalOutput(TestDvc):
def _get_cls(self):
return LocalOutput
def _get_output(self):
stage = Stage(self.dvc)
return self._get_cls()(stage, "path", cache=False)
def test_save_missing(self):
o = self._get_output()
with patch.object(o.fs, "exists", return_value=False):
with self.assertRaises(o.DoesNotExistError):
o.save()
def test_str_workdir_outside_repo(tmp_dir, erepo_dir):
stage = Stage(erepo_dir.dvc)
output = LocalOutput(stage, "path", cache=False)
output = Output(stage, "path", cache=False)
assert relpath("path", erepo_dir.dvc.root_dir) == str(output)
def test_str_workdir_inside_repo(dvc):
stage = Stage(dvc)
output = LocalOutput(stage, "path", cache=False)
output = Output(stage, "path", cache=False)
assert "path" == str(output)
stage = Stage(dvc, wdir="some_folder")
output = LocalOutput(stage, "path", cache=False)
output = Output(stage, "path", cache=False)
assert os.path.join("some_folder", "path") == str(output)
@@ -48,7 +33,7 @@ def test_str_on_local_absolute_path(dvc):
rel_path = os.path.join("path", "to", "file")
abs_path = os.path.abspath(rel_path)
output = LocalOutput(stage, abs_path, cache=False)
output = Output(stage, abs_path, cache=False)
assert output.def_path == rel_path
assert output.path_info.fspath == abs_path
@@ -60,7 +45,7 @@ def test_str_on_external_absolute_path(dvc):
rel_path = os.path.join("..", "path", "to", "file")
abs_path = os.path.abspath(rel_path)
output = LocalOutput(stage, abs_path, cache=False)
output = Output(stage, abs_path, cache=False)
assert output.def_path == abs_path
assert output.path_info.fspath == abs_path
@@ -70,7 +55,7 @@ def test_str_on_external_absolute_path(dvc):
class TestGetFilesNumber(TestDvc):
def _get_output(self):
stage = Stage(self.dvc)
return LocalOutput(stage, "path")
return Output(stage, "path")
def test_return_0_on_no_cache(self):
o = self._get_output()
@@ -82,7 +67,7 @@ class TestGetFilesNumber(TestDvc):
o.hash_info = HashInfo("md5", "12345678.dir", nfiles=2)
self.assertEqual(2, o.get_files_number())
@patch.object(LocalOutput, "is_dir_checksum", False)
@patch.object(Output, "is_dir_checksum", False)
def test_return_1_on_single_file_cache(self):
o = self._get_output()
o.hash_info = HashInfo("md5", "12345678")
tests/unit/output/test_output.py
+18 -8
View File
@@ -1,4 +1,5 @@
import logging
import os
import pytest
from funcy import first
@@ -6,7 +7,17 @@ from voluptuous import MultipleInvalid, Schema
from dvc.ignore import _no_match
from dvc.objects.db import NamedCache
from dvc.output import CHECKSUM_SCHEMA, BaseOutput
from dvc.output import CHECKSUM_SCHEMA, Output
from dvc.path_info import PathInfo
from dvc.stage import Stage
def test_save_missing(dvc, mocker):
stage = Stage(dvc)
out = Output(stage, "path", cache=False)
with mocker.patch.object(out.fs, "exists", return_value=False):
with pytest.raises(out.DoesNotExistError):
out.save()
@pytest.mark.parametrize(
@@ -73,20 +84,19 @@ def test_get_used_cache(exists, expected_message, mocker, caplog):
stage = mocker.MagicMock()
mocker.patch.object(stage, "__str__", return_value="stage: 'stage.dvc'")
mocker.patch.object(stage, "addressing", "stage.dvc")
mocker.patch.object(stage, "wdir", ".")
mocker.patch.object(stage, "wdir", PathInfo(os.getcwd()))
mocker.patch.object(stage.repo, "root_dir", os.getcwd())
mocker.patch.object(stage.repo.dvcignore, "is_ignored", return_value=False)
mocker.patch.object(
stage.repo.fs.dvcignore, "is_ignored", return_value=False
)
mocker.patch.object(
stage.repo.fs.dvcignore, "check_ignore", return_value=_no_match("path")
stage.repo.dvcignore, "check_ignore", return_value=_no_match("path")
)
output = BaseOutput(stage, "path")
output = Output(stage, "path")
mocker.patch.object(output, "use_cache", True)
mocker.patch.object(stage, "is_repo_import", False)
mocker.patch.object(
BaseOutput, "exists", new_callable=mocker.PropertyMock
Output, "exists", new_callable=mocker.PropertyMock
).return_value = exists
with caplog.at_level(logging.WARNING, logger="dvc"):
tests/unit/output/test_s3.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.output.s3 import S3Output
from tests.unit.output.test_local import TestLocalOutput
class TestS3Output(TestLocalOutput):
def _get_cls(self):
return S3Output
tests/unit/output/test_ssh.py
-7
View File
@@ -1,7 +0,0 @@
from dvc.output.ssh import SSHOutput
from tests.unit.output.test_local import TestLocalOutput
class TestSSHOutput(TestLocalOutput):
def _get_cls(self):
return SSHOutput
tests/unit/stage/test_cache.py
+4 -4
View File
@@ -2,7 +2,7 @@ import os
import pytest
from dvc.output import base
import dvc.output as dvc_output
def test_stage_cache(tmp_dir, dvc, mocker):
@@ -44,7 +44,7 @@ def test_stage_cache(tmp_dir, dvc, mocker):
assert os.path.isfile(cache_file)
run_spy = mocker.patch("dvc.stage.run.cmd_run")
checkout_spy = mocker.spy(base, "checkout")
checkout_spy = mocker.spy(dvc_output, "checkout")
with dvc.lock:
stage.run()
@@ -97,7 +97,7 @@ def test_stage_cache_params(tmp_dir, dvc, mocker):
assert os.path.isfile(cache_file)
run_spy = mocker.patch("dvc.stage.run.cmd_run")
checkout_spy = mocker.spy(base, "checkout")
checkout_spy = mocker.spy(dvc_output, "checkout")
with dvc.lock:
stage.run()
@@ -151,7 +151,7 @@ def test_stage_cache_wdir(tmp_dir, dvc, mocker):
assert os.path.isfile(cache_file)
run_spy = mocker.patch("dvc.stage.run.cmd_run")
checkout_spy = mocker.spy(base, "checkout")
checkout_spy = mocker.spy(dvc_output, "checkout")
with dvc.lock:
stage.run()