Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def git_object_by_path(self, path):
import git
path = relpath(os.path.realpath(path), self.git.working_dir)
assert path.split(os.sep, 1)[0] != ".."
try:
tree = self.git.tree(self.rev)
except git.exc.BadName as exc:
raise DvcException(
"revision '{}' not found in git '{}'".format(
self.rev, os.path.relpath(self.git.working_dir)
),
cause=exc,
)
if not path or path == ".":
return tree
for i in path.split(os.sep):
if not self._is_tree_and_contains(tree, i):
# there is no tree for specified path
return None
tree = tree[i]
return tree
def makedirs(self, path):
# Single stat call will say whether this is a dir, a file or a link
st_mode = self.st_mode(path)
if stat.S_ISDIR(st_mode):
return
if stat.S_ISREG(st_mode) or stat.S_ISLNK(st_mode):
raise DvcException(
"a file with the same name '{}' already exists".format(path)
)
head, tail = posixpath.split(path)
if head:
self.makedirs(head)
if tail:
try:
self.sftp.mkdir(path)
except IOError as e:
# Since paramiko errors are very vague we need to recheck
# whether it's because path already exists or something else
if e.errno == errno.EACCES or not self.exists(path):
raise DvcException(
host=path_info["host"],
user=path_info["user"],
port=path_info["port"],
)
# Use different md5 commands depending on os
stdout = self._exec(ssh, "uname").strip()
if stdout == "Darwin":
md5cmd = "md5"
index = -1
elif stdout == "Linux":
md5cmd = "md5sum"
index = 0
else:
msg = "'{}' is not supported as a remote".format(stdout)
raise DvcException(msg)
stdout = self._exec(ssh, "{} {}".format(md5cmd, path_info["path"]))
md5 = stdout.split()[index]
ssh.close()
assert len(md5) == 32
return md5
"the following stages:"
"\n"
"{stages}".format(stages=stages)
)
super(CyclicGraphError, self).__init__(msg)
class ConfirmRemoveError(DvcException):
def __init__(self, path):
super(ConfirmRemoveError, self).__init__(
"unable to remove '{}' without a confirmation from the user. Use "
"`-f` to force.".format(path)
)
class InitError(DvcException):
def __init__(self, msg):
super(InitError, self).__init__(msg)
class ReproductionError(DvcException):
def __init__(self, dvc_file_name, ex):
self.path = dvc_file_name
msg = "failed to reproduce '{}'".format(dvc_file_name)
super(ReproductionError, self).__init__(msg, cause=ex)
class BadMetricError(DvcException):
def __init__(self, paths):
super(BadMetricError, self).__init__(
"the following metrics do not exists, "
"are not metric files or are malformed: {paths}".format(
from dvc.exceptions import DvcException
from dvc.progress import Tqdm
from dvc.system import System
src = fspath_py35(src)
dest = fspath_py35(dest)
name = name if name else os.path.basename(dest)
total = os.stat(src).st_size
if os.path.isdir(dest):
dest = os.path.join(dest, os.path.basename(src))
try:
System.reflink(src, dest)
except DvcException:
with Tqdm(
desc=name, disable=no_progress_bar, total=total, bytes=True
) as pbar:
with open(src, "rb") as fsrc, open(dest, "wb+") as fdest:
while True:
buf = fsrc.read(LOCAL_CHUNK_SIZE)
if not buf:
break
fdest.write(buf)
pbar.update(len(buf))
"the following stages:"
"\n"
"{stages}".format(stages=stages)
)
super(CyclicGraphError, self).__init__(msg)
class ConfirmRemoveError(DvcException):
def __init__(self, path):
super(ConfirmRemoveError, self).__init__(
"unable to remove '{}' without a confirmation from the user. Use "
"'-f' to force.".format(path)
)
class InitError(DvcException):
def __init__(self, msg):
super(InitError, self).__init__(msg)
class ReproductionError(DvcException):
def __init__(self, dvc_file_name, ex):
self.path = dvc_file_name
msg = "failed to reproduce '{}'".format(dvc_file_name)
super(ReproductionError, self).__init__(msg, cause=ex)
class BadMetricError(DvcException):
def __init__(self, paths):
super(BadMetricError, self).__init__(
"the following metrics do not exists, "
"are not metric files or are malformed: {paths}".format(
def __init__(self, path, cause=None):
super(FileMissingError, self).__init__(
"Can't find '{}' neither locally nor on remote".format(path),
cause=cause,
)
class DvcIgnoreInCollectedDirError(DvcException):
def __init__(self, ignore_dirname):
super(DvcIgnoreInCollectedDirError, self).__init__(
".dvcignore file should not be in collected dir path: "
"'{}'".format(ignore_dirname)
)
class UrlNotDvcRepoError(DvcException):
def __init__(self, url):
super(UrlNotDvcRepoError, self).__init__(
"URL '{}' is not a dvc repository.".format(url)
)
class GetDVCFileError(DvcException):
def __init__(self):
super(GetDVCFileError, self).__init__(
"the given path is a DVC-file, you must specify a data file "
"or a directory"
)
class GitHookAlreadyExistsError(DvcException):
def __init__(self, hook_name):
self.path = dvc_file_name
msg = "failed to reproduce '{}'".format(dvc_file_name)
super(ReproductionError, self).__init__(msg, cause=ex)
class BadMetricError(DvcException):
def __init__(self, paths):
super(BadMetricError, self).__init__(
"the following metrics do not exists, "
"are not metric files or are malformed: {paths}".format(
paths=", ".join("'{}'".format(path) for path in paths)
)
)
class NoMetricsError(DvcException):
def __init__(self):
super(NoMetricsError, self).__init__(
"no metric files in this repository. "
"Use `dvc metrics add` to add a metric file to track."
)
class StageFileCorruptedError(DvcException):
def __init__(self, path, cause=None):
path = relpath(path)
super(StageFileCorruptedError, self).__init__(
"unable to read DVC-file: {} "
"YAML file structure is corrupted".format(path),
cause=cause,
)
def _try_links(self, from_info, to_info, link_types):
while link_types:
link_method = getattr(self, link_types[0])
try:
self._do_link(from_info, to_info, link_method)
self.cache_type_confirmed = True
return
except DvcException as exc:
msg = "Cache type '{}' is not supported: {}"
logger.debug(msg.format(link_types[0], str(exc)))
del link_types[0]
raise DvcException("no possible cache types left to try out.")
import logging
import os
import re
import configobj
from voluptuous import Schema, Required, Optional, Invalid
from voluptuous import All, Any, Lower, Range, Coerce, Match
from dvc.exceptions import DvcException
from dvc.exceptions import NotDvcRepoError
from dvc.utils.compat import open, str
logger = logging.getLogger(__name__)
class ConfigError(DvcException):
"""DVC config exception.
Args:
msg (str): error message.
cause (Exception): optional exception that has caused this error.
"""
def __init__(self, msg, cause=None):
super(ConfigError, self).__init__(
"config file error: {}".format(msg), cause=cause
)
class NoRemoteError(ConfigError):
def __init__(self, command, cause=None):
msg = (