mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-09 23:10:24 +01:00
skshetry
b647475a64
* drop Python 3.8 Note that this PR only makes typing related changes. Other changes can be done once this gets merged, they are relatively few in numbers. * bump build python version * add typing_extensions to lint group
59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
from dvc.exceptions import DvcException
|
|
from dvc.fs import download as fs_download
|
|
from dvc.output import Output
|
|
|
|
|
|
class DependencyDoesNotExistError(DvcException):
|
|
def __init__(self, path):
|
|
msg = f"dependency '{path}' does not exist"
|
|
super().__init__(msg)
|
|
|
|
|
|
class DependencyIsNotFileOrDirError(DvcException):
|
|
def __init__(self, path):
|
|
msg = f"dependency '{path}' is not a file or directory"
|
|
super().__init__(msg)
|
|
|
|
|
|
class DependencyIsStageFileError(DvcException):
|
|
def __init__(self, path):
|
|
super().__init__(f"DVC file '{path}' cannot be a dependency.")
|
|
|
|
|
|
class Dependency(Output):
|
|
IS_DEPENDENCY = True
|
|
|
|
DoesNotExistError: type[DvcException] = DependencyDoesNotExistError
|
|
IsNotFileOrDirError: type[DvcException] = DependencyIsNotFileOrDirError
|
|
IsStageFileError: type[DvcException] = DependencyIsStageFileError
|
|
|
|
def workspace_status(self) -> dict[str, str]:
|
|
if self.fs.version_aware:
|
|
old_fs_path = self.fs_path
|
|
try:
|
|
self.fs_path = self.fs.version_path(self.fs_path, None)
|
|
if self.changed_meta():
|
|
return {str(self): "update available"}
|
|
finally:
|
|
self.fs_path = old_fs_path
|
|
return super().workspace_status()
|
|
|
|
def update(self, rev=None):
|
|
if self.fs.version_aware:
|
|
self.fs_path = self.fs.version_path(self.fs_path, rev)
|
|
self.meta = self.get_meta()
|
|
self.fs_path = self.fs.version_path(self.fs_path, self.meta.version_id)
|
|
|
|
def download(self, to, jobs=None):
|
|
fs_download(self.fs, self.fs_path, to.fs_path, jobs=jobs)
|
|
|
|
def save(self):
|
|
super().save()
|
|
if self.fs.version_aware:
|
|
self.fs_path = self.fs.version_path(self.fs_path, self.meta.version_id)
|
|
|
|
def dumpd(self, **kwargs):
|
|
if self.fs.version_aware:
|
|
kwargs["with_files"] = True
|
|
return super().dumpd(**kwargs)
|