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

dvc.fs: move fs to dvc.objects.fs

Browse Source 
Some of the APIs/FileSystems are reimported inside dvc.fs module.
This commit is contained in:
Saugat Pachhai (सौगात)
2022-05-16 13:15:19 +03:00
committed by Ruslan Kuprieiev
parent b8e770747f
commit 570e3a5359
95 changed files with 552 additions and 521 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/config.py
+2 -2
View File
@@ -86,7 +86,7 @@ class Config(dict):
def __init__(
self, dvc_dir=None, validate=True, fs=None, config=None
): # pylint: disable=super-init-not-called
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
self.dvc_dir = dvc_dir
self.wfs = LocalFileSystem()
@@ -227,7 +227,7 @@ class Config(dict):
@staticmethod
def _to_relpath(conf_dir, path):
from dvc.fs.local import localfs
from dvc.fs import localfs
from dvc.utils import relpath
from .config_schema import RelPath
dvc/data/checkout.py
+2 -2
View File
@@ -3,9 +3,9 @@ from itertools import chain
from dvc import prompt
from dvc.exceptions import CacheLinkError, CheckoutError, ConfirmRemoveError
from dvc.fs._callback import FsspecCallback
from dvc.fs.generic import test_links, transfer
from dvc.ignore import DvcIgnoreFilter
from dvc.objects.fs._callback import FsspecCallback
from dvc.objects.fs.generic import test_links, transfer
from dvc.types import Optional
from .diff import ROOT
dvc/data/db/__init__.py
+1 -1
View File
@@ -18,7 +18,7 @@ def get_odb(fs, fs_path, **config):
def _get_odb(repo, settings):
from dvc.fs import get_cloud_fs
from dvc.objects.fs import get_cloud_fs
if not settings:
return None
dvc/data/db/index.py
+2 -2
View File
@@ -88,9 +88,9 @@ class ObjectDBIndex(ObjectDBIndexBase):
tmp_dir: "StrPath",
name: str,
): # pylint: disable=super-init-not-called
from dvc.fs.local import LocalFileSystem
from dvc.fs.utils import makedirs
from dvc.objects.cache import Cache, Index
from dvc.objects.fs.local import LocalFileSystem
from dvc.objects.fs.utils import makedirs
self.index_dir = os.path.join(tmp_dir, self.INDEX_DIR, name)
makedirs(self.index_dir, exist_ok=True)
dvc/data/db/local.py
+3 -3
View File
@@ -6,10 +6,10 @@ from functools import partial
from funcy import cached_property
from shortuuid import uuid
from dvc.fs.system import umask
from dvc.fs.utils import copyfile, relpath, remove, walk_files
from dvc.objects.db import ObjectDB, noop, wrap_iter
from dvc.objects.errors import ObjectFormatError
from dvc.objects.fs.system import umask
from dvc.objects.fs.utils import copyfile, relpath, remove, walk_files
from dvc.objects.hash_info import HashInfo
logger = logging.getLogger(__name__)
@@ -49,7 +49,7 @@ class LocalObjectDB(ObjectDB):
os.chmod(to_info, self._file_mode)
def makedirs(self, fs_path):
from dvc.fs.utils import makedirs
from dvc.objects.fs.utils import makedirs
makedirs(fs_path, exist_ok=True, mode=self._dir_mode)
dvc/data/db/reference.py
+2 -2
View File
@@ -9,8 +9,8 @@ from dvc.scheme import Schemes
from ..reference import ReferenceHashFile
if TYPE_CHECKING:
from dvc.fs._callback import FsspecCallback
from dvc.fs.base import AnyFSPath, FileSystem
from dvc.objects.fs._callback import FsspecCallback
from dvc.objects.fs.base import AnyFSPath, FileSystem
from dvc.objects.hash_info import HashInfo
logger = logging.getLogger(__name__)
dvc/data/reference.py
+2 -2
View File
@@ -8,8 +8,8 @@ from dvc.objects.errors import ObjectFormatError
from dvc.objects.file import HashFile
if TYPE_CHECKING:
from dvc.fs.base import AnyFSPath, FileSystem
from dvc.objects.db import ObjectDB
from dvc.objects.fs.base import AnyFSPath, FileSystem
from dvc.objects.hash_info import HashInfo
logger = logging.getLogger(__name__)
@@ -70,8 +70,8 @@ class ReferenceHashFile(HashFile):
@classmethod
def from_bytes(cls, data: bytes, fs_cache: Optional[dict] = None):
from dvc.fs import get_fs_cls
from dvc.fs.dvc import DvcFileSystem, _DvcFileSystem
from dvc.objects.fs import get_fs_cls
try:
dict_ = pickle.loads(data)
dvc/data/stage.py
+4 -4
View File
@@ -17,8 +17,8 @@ from dvc.progress import Tqdm
from .db.reference import ReferenceObjectDB
if TYPE_CHECKING:
from dvc.fs.base import AnyFSPath, FileSystem
from dvc.objects.db import ObjectDB
from dvc.objects.fs.base import AnyFSPath, FileSystem
from .tree import Tree
@@ -30,8 +30,8 @@ _STAGING_MEMFS_PATH = "dvc-staging"
def _upload_file(from_fs_path, fs, odb, upload_odb, callback=None):
from dvc.fs._callback import FsspecCallback
from dvc.fs.utils import tmp_fname
from dvc.objects.fs._callback import FsspecCallback
from dvc.objects.fs.utils import tmp_fname
from .stream import HashedStreamReader
@@ -194,7 +194,7 @@ def _get_staging(odb: "ObjectDB") -> "ReferenceObjectDB":
Staging will be a reference ODB stored in the the global memfs.
"""
from dvc.fs.memory import MemoryFileSystem
from dvc.objects.fs.memory import MemoryFileSystem
fs = MemoryFileSystem()
fs_path = _make_staging_url(fs, odb, odb.fs_path)
dvc/data/tree.py
+2 -2
View File
@@ -55,8 +55,8 @@ class Tree(HashFile):
self._dict[key] = (meta, oid)
def digest(self, hash_info: Optional["HashInfo"] = None):
from dvc.fs.memory import MemoryFileSystem
from dvc.fs.utils import tmp_fname
from dvc.objects.fs.memory import MemoryFileSystem
from dvc.objects.fs.utils import tmp_fname
memfs = MemoryFileSystem()
fs_path = "memory://{}".format(tmp_fname(""))
dvc/dvcfile.py
+1 -1
View File
@@ -59,7 +59,7 @@ def is_lock_file(path):
def is_git_ignored(repo, path):
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
from dvc.scm import NoSCMError
try:
dvc/external_repo.py
+2 -2
View File
@@ -33,7 +33,7 @@ def external_repo(
from scmrepo.git import Git
from dvc.config import NoRemoteError
from dvc.fs.git import GitFileSystem
from dvc.fs import GitFileSystem
logger.debug("Creating external repo %s@%s", url, rev)
path = _cached_clone(url, rev, for_write=for_write)
@@ -95,7 +95,7 @@ def external_repo(
def erepo_factory(url, root_dir, cache_config):
from dvc.fs.local import localfs
from dvc.fs import localfs
def make_repo(path, fs=None, **_kwargs):
_config = cache_config.copy()
dvc/fs/__init__.py
+37 -125
View File
@@ -1,126 +1,38 @@
from urllib.parse import urlparse
# pylint: disable=unused-import
from dvc.objects.fs import utils # noqa: F401
from dvc.objects.fs import ( # noqa: F401
FS_MAP,
AzureFileSystem,
GDriveFileSystem,
GSFileSystem,
HDFSFileSystem,
HTTPFileSystem,
HTTPSFileSystem,
LocalFileSystem,
OSSFileSystem,
S3FileSystem,
SSHFileSystem,
WebDAVFileSystem,
WebDAVSFileSystem,
WebHDFSFileSystem,
_resolve_remote_refs,
generic,
get_cloud_fs,
get_fs_cls,
get_fs_config,
system,
)
from dvc.objects.fs.azure import AzureAuthError # noqa: F401
from dvc.objects.fs.base import ( # noqa: F401
AnyFSPath,
FileSystem,
RemoteMissingDepsError,
)
from dvc.objects.fs.gdrive import GDriveAuthError # noqa: F401
from dvc.objects.fs.git import GitFileSystem # noqa: F401
from dvc.objects.fs.local import localfs # noqa: F401
from dvc.objects.fs.memory import MemoryFileSystem # noqa: F401
from dvc.objects.fs.path import Path # noqa: F401
from dvc.objects.fs.ssh import DEFAULT_PORT as DEFAULT_SSH_PORT # noqa: F401
from ..scheme import Schemes
from . import generic # noqa: F401
from .azure import AzureFileSystem
from .gdrive import GDriveFileSystem
from .gs import GSFileSystem
from .hdfs import HDFSFileSystem
from .http import HTTPFileSystem
from .https import HTTPSFileSystem
from .local import LocalFileSystem
from .oss import OSSFileSystem
from .s3 import S3FileSystem
from .ssh import SSHFileSystem
from .webdav import WebDAVFileSystem, WebDAVSFileSystem
from .webhdfs import WebHDFSFileSystem
FS_MAP = {
Schemes.AZURE: AzureFileSystem,
Schemes.GDRIVE: GDriveFileSystem,
Schemes.GS: GSFileSystem,
Schemes.HDFS: HDFSFileSystem,
Schemes.WEBHDFS: WebHDFSFileSystem,
Schemes.HTTP: HTTPFileSystem,
Schemes.HTTPS: HTTPSFileSystem,
Schemes.S3: S3FileSystem,
Schemes.SSH: SSHFileSystem,
Schemes.OSS: OSSFileSystem,
Schemes.WEBDAV: WebDAVFileSystem,
Schemes.WEBDAVS: WebDAVSFileSystem,
# NOTE: LocalFileSystem is the default
}
def get_fs_cls(remote_conf, scheme=None):
if not scheme:
scheme = urlparse(remote_conf["url"]).scheme
return FS_MAP.get(scheme, LocalFileSystem)
def get_fs_config(repo, config, **kwargs):
name = kwargs.get("name")
if name:
try:
remote_conf = config["remote"][name.lower()]
except KeyError:
from dvc.config import RemoteNotFoundError
raise RemoteNotFoundError(f"remote '{name}' doesn't exist")
else:
remote_conf = kwargs
return _resolve_remote_refs(repo, config, remote_conf)
def _resolve_remote_refs(repo, config, remote_conf):
# Support for cross referenced remotes.
# This will merge the settings, shadowing base ref with remote_conf.
# For example, having:
#
# dvc remote add server ssh://localhost
# dvc remote modify server user root
# dvc remote modify server ask_password true
#
# dvc remote add images remote://server/tmp/pictures
# dvc remote modify images user alice
# dvc remote modify images ask_password false
# dvc remote modify images password asdf1234
#
# Results on a config dictionary like:
#
# {
# "url": "ssh://localhost/tmp/pictures",
# "user": "alice",
# "password": "asdf1234",
# "ask_password": False,
# }
parsed = urlparse(remote_conf["url"])
if parsed.scheme != "remote":
return remote_conf
base = get_fs_config(repo, config, name=parsed.netloc)
cls, _, _ = get_cloud_fs(repo, **base)
relpath = parsed.path.lstrip("/").replace("/", cls.sep)
url = cls.sep.join((base["url"], relpath))
return {**base, **remote_conf, "url": url}
def get_cloud_fs(repo, **kwargs):
from dvc.config import ConfigError
from dvc.config_schema import SCHEMA, Invalid
repo_config = repo.config if repo else {}
core_config = repo_config.get("core", {})
remote_conf = get_fs_config(repo, 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 = core_config.get("jobs")
if jobs:
remote_conf["jobs"] = jobs
if "checksum_jobs" not in remote_conf:
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 and repo:
remote_conf["gdrive_credentials_tmp_dir"] = repo.tmp_dir
url = remote_conf.pop("url")
if issubclass(cls, WebDAVFileSystem):
# For WebDAVFileSystem, provided url is the base path itself, so it
# should be treated as being a root path.
fs_path = cls.root_marker
else:
fs_path = cls._strip_protocol(url) # pylint:disable=protected-access
extras = cls._get_kwargs_from_urls(url) # pylint:disable=protected-access
conf = {**extras, **remote_conf} # remote config takes priority
return cls, conf, fs_path
# pylint: enable=unused-import
dvc/fs/_callback.py
+7 -235
View File
@@ -1,235 +1,7 @@
from contextlib import ExitStack
from functools import wraps
from typing import TYPE_CHECKING, Any, Dict, Optional, TypeVar, overload
import fsspec
from funcy import cached_property
if TYPE_CHECKING:
from typing import BinaryIO, Callable, TextIO, Union
from typing_extensions import ParamSpec
from dvc.progress import Tqdm
from dvc.ui._rich_progress import RichTransferProgress
_P = ParamSpec("_P")
_R = TypeVar("_R")
class FsspecCallback(fsspec.Callback):
"""FsspecCallback usable as a context manager, and a few helper methods."""
@overload
def wrap_attr(self, fobj: "BinaryIO", method: str = "read") -> "BinaryIO":
...
@overload
def wrap_attr(self, fobj: "TextIO", method: str = "read") -> "TextIO":
...
def wrap_attr(
self, fobj: "Union[TextIO, BinaryIO]", method: str = "read"
) -> "Union[TextIO, BinaryIO]":
from tqdm.utils import CallbackIOWrapper
wrapped = CallbackIOWrapper(self.relative_update, fobj, method)
return wrapped
def wrap_fn(self, fn: "Callable[_P, _R]") -> "Callable[_P, _R]":
@wraps(fn)
def wrapped(*args: "_P.args", **kwargs: "_P.kwargs") -> "_R":
res = fn(*args, **kwargs)
self.relative_update()
return res
return wrapped
def wrap_and_branch(self, fn: "Callable") -> "Callable":
"""
Wraps a function, and pass a new child callback to it.
When the function completes, we increment the parent callback by 1.
"""
wrapped = self.wrap_fn(fn)
@wraps(fn)
def func(path1: str, path2: str):
kw: Dict[str, Any] = {}
with self.branch(path1, path2, kw):
return wrapped(path1, path2, **kw)
return func
def __enter__(self):
return self
def __exit__(self, *exc_args):
self.close()
def close(self):
"""Handle here on exit."""
def relative_update(self, inc: int = 1) -> None:
inc = inc if inc is not None else 0
return super().relative_update(inc)
def absolute_update(self, value: int) -> None:
value = value if value is not None else self.value
return super().absolute_update(value)
@classmethod
def as_callback(
cls, maybe_callback: Optional["FsspecCallback"] = None
) -> "FsspecCallback":
if maybe_callback is None:
return DEFAULT_CALLBACK
return maybe_callback
@classmethod
def as_tqdm_callback(
cls,
callback: Optional["FsspecCallback"] = None,
**tqdm_kwargs: Any,
) -> "FsspecCallback":
return callback or TqdmCallback(**tqdm_kwargs)
@classmethod
def as_rich_callback(
cls, callback: Optional["FsspecCallback"] = None, **rich_kwargs
):
return callback or RichCallback(**rich_kwargs)
def branch(
self,
path_1: str,
path_2: str,
kwargs: Dict[str, Any],
child: "FsspecCallback" = None,
) -> "FsspecCallback":
child = kwargs["callback"] = child or DEFAULT_CALLBACK
return child
class NoOpCallback(FsspecCallback, fsspec.callbacks.NoOpCallback):
pass
class TqdmCallback(FsspecCallback):
def __init__(
self,
size: Optional[int] = None,
value: int = 0,
progress_bar: "Tqdm" = None,
**tqdm_kwargs,
):
tqdm_kwargs["total"] = size or -1
self._tqdm_kwargs = tqdm_kwargs
self._progress_bar = progress_bar
self._stack = ExitStack()
super().__init__(size=size, value=value)
@cached_property
def progress_bar(self):
from dvc.progress import Tqdm
progress_bar = (
self._progress_bar
if self._progress_bar is not None
else Tqdm(**self._tqdm_kwargs)
)
return self._stack.enter_context(progress_bar)
def __enter__(self):
return self
def close(self):
self._stack.close()
def set_size(self, size):
# Tqdm tries to be smart when to refresh,
# so we try to force it to re-render.
super().set_size(size)
self.progress_bar.refresh()
def call(self, hook_name=None, **kwargs):
self.progress_bar.update_to(self.value, total=self.size)
def branch(
self,
path_1: str,
path_2: str,
kwargs,
child: Optional[FsspecCallback] = None,
):
child = child or TqdmCallback(bytes=True, desc=path_1)
return super().branch(path_1, path_2, kwargs, child=child)
class RichCallback(FsspecCallback):
def __init__(
self,
size: Optional[int] = None,
value: int = 0,
progress: "RichTransferProgress" = None,
desc: str = None,
bytes: bool = False, # pylint: disable=redefined-builtin
unit: str = None,
disable: bool = False,
) -> None:
self._progress = progress
self.disable = disable
self._task_kwargs = {
"description": desc or "",
"bytes": bytes,
"unit": unit,
"total": size or 0,
"visible": False,
"progress_type": None if bytes else "summary",
}
self._stack = ExitStack()
super().__init__(size=size, value=value)
@cached_property
def progress(self):
from dvc.ui import ui
from dvc.ui._rich_progress import RichTransferProgress
if self._progress is not None:
return self._progress
progress = RichTransferProgress(
transient=True,
disable=self.disable,
console=ui.error_console,
)
return self._stack.enter_context(progress)
@cached_property
def task(self):
return self.progress.add_task(**self._task_kwargs)
def __enter__(self):
return self
def close(self):
self.progress.clear_task(self.task)
self._stack.close()
def call(self, hook_name=None, **kwargs):
self.progress.update(
self.task,
completed=self.value,
total=self.size,
visible=not self.disable,
)
def branch(
self, path_1, path_2, kwargs, child: Optional[FsspecCallback] = None
):
child = child or RichCallback(
progress=self.progress, desc=path_1, bytes=True
)
return super().branch(path_1, path_2, kwargs, child=child)
DEFAULT_CALLBACK = NoOpCallback()
# pylint: disable=unused-import
from dvc.objects.fs._callback import ( # noqa: F401
DEFAULT_CALLBACK,
FsspecCallback,
RichCallback,
TqdmCallback,
)
dvc/fs/data.py
+3 -3
View File
@@ -5,8 +5,8 @@ import typing
from fsspec import AbstractFileSystem
from funcy import cached_property, wrap_prop
from ._callback import DEFAULT_CALLBACK
from .base import FileSystem
from dvc.objects.fs._callback import DEFAULT_CALLBACK
from dvc.objects.fs.base import FileSystem
if typing.TYPE_CHECKING:
from dvc.types import AnyPath
@@ -30,7 +30,7 @@ class _DataFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
@cached_property
def path(self):
from .path import Path
from . import Path
def _getcwd():
return self.root_marker
dvc/fs/dvc.py
+12 -7
View File
@@ -1,5 +1,7 @@
import logging
import ntpath
import os
import posixpath
import threading
from contextlib import suppress
from itertools import takewhile
@@ -8,10 +10,11 @@ from typing import TYPE_CHECKING, Callable, Optional, Tuple, Type, Union
from fsspec.spec import AbstractFileSystem
from funcy import cached_property, wrap_prop, wrap_with
from ._callback import DEFAULT_CALLBACK
from .base import FileSystem
from dvc.objects.fs._callback import DEFAULT_CALLBACK
from dvc.objects.fs.base import FileSystem
from dvc.objects.fs.path import Path
from .data import DataFileSystem
from .path import Path
if TYPE_CHECKING:
from dvc.repo import Repo
@@ -21,6 +24,10 @@ logger = logging.getLogger(__name__)
RepoFactory = Union[Callable[[str], "Repo"], Type["Repo"]]
def as_posix(path: str) -> str:
return path.replace(ntpath.sep, posixpath.sep)
def _wrap_walk(dvc_fs, *args, **kwargs):
for root, dnames, fnames in dvc_fs.walk(*args, **kwargs):
yield dvc_fs.path.join(dvc_fs.repo.root_dir, root), dnames, fnames
@@ -41,7 +48,7 @@ def _ls(fs, path):
def _merge_info(repo, fs_info, dvc_info):
from dvc.fs.utils import is_exec
from . import utils
ret = {"repo": repo}
@@ -57,7 +64,7 @@ def _merge_info(repo, fs_info, dvc_info):
ret["size"] = fs_info["size"]
isexec = False
if fs_info["type"] == "file":
isexec = is_exec(fs_info["mode"])
isexec = utils.is_exec(fs_info["mode"])
ret["isexec"] = isexec
return ret
@@ -468,8 +475,6 @@ class DvcFileSystem(FileSystem):
return self.fs.config
def from_os_path(self, path):
from .utils import as_posix
if os.path.isabs(path):
path = os.path.relpath(path, self.repo.root_dir)
dvc/ignore.py
+1 -2
View File
@@ -8,8 +8,7 @@ from pathspec.patterns import GitWildMatchPattern
from pathspec.util import normalize_file
from pygtrie import Trie
from dvc.fs.base import AnyFSPath, FileSystem
from dvc.fs.local import localfs
from dvc.fs import AnyFSPath, FileSystem, localfs
from dvc.pathspec_math import PatternInfo, merge_patterns
from dvc.scheme import Schemes
from dvc.types import List, Optional
dvc/info.py
+2 -3
View File
@@ -7,8 +7,7 @@ import psutil
from dvc import __version__
from dvc.exceptions import NotDvcRepoError
from dvc.fs import FS_MAP, get_fs_cls, get_fs_config
from dvc.fs.generic import test_links
from dvc.fs import FS_MAP, generic, get_fs_cls, get_fs_config
from dvc.repo import Repo
from dvc.scm import SCMError
from dvc.utils import error_link
@@ -83,7 +82,7 @@ def _get_remotes(config):
def _get_linktype_support_info(repo):
odb = repo.odb.local
links = test_links(
links = generic.test_links(
["reflink", "hardlink", "symlink"],
odb.fs,
odb.fs_path,
dvc/machine/backend/base.py
+1 -1
View File
@@ -3,7 +3,7 @@ from contextlib import contextmanager
from typing import TYPE_CHECKING, Iterator, Optional
if TYPE_CHECKING:
from dvc.fs.ssh import SSHFileSystem
from dvc.fs import SSHFileSystem
from dvc.types import StrPath
dvc/machine/backend/terraform.py
+1 -2
View File
@@ -4,8 +4,7 @@ from functools import partial, partialmethod
from typing import TYPE_CHECKING, Iterator, Optional
from dvc.exceptions import DvcException
from dvc.fs.ssh import DEFAULT_PORT as DEFAULT_SSH_PORT
from dvc.fs.ssh import SSHFileSystem
from dvc.fs import DEFAULT_SSH_PORT, SSHFileSystem
from dvc.utils.fs import makedirs
from .base import BaseMachineBackend
dvc/objects/db.py
+6 -7
View File
@@ -12,9 +12,8 @@ from .file import HashFile
if TYPE_CHECKING:
from typing import Tuple
from dvc.fs._callback import FsspecCallback
from dvc.fs.base import AnyFSPath, FileSystem
from .fs._callback import FsspecCallback
from .fs.base import AnyFSPath, FileSystem
from .hash_info import HashInfo
logger = logging.getLogger(__name__)
@@ -95,11 +94,11 @@ class ObjectDB:
hardlink: bool = False,
callback: "FsspecCallback" = None,
):
from dvc import fs
from dvc.fs._callback import FsspecCallback
from .fs import generic
from .fs._callback import FsspecCallback
self.makedirs(self.fs.path.parent(to_info))
return fs.generic.transfer(
return generic.transfer(
from_fs,
from_info,
self.fs,
@@ -117,7 +116,7 @@ class ObjectDB:
verify: Optional[bool] = None,
callback: "FsspecCallback" = None,
):
from dvc.fs._callback import FsspecCallback
from .fs._callback import FsspecCallback
if self.read_only:
raise ObjectDBPermissionError("Cannot add to read-only ODB")
dvc/objects/file.py
+1 -2
View File
@@ -4,9 +4,8 @@ import os
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from dvc.fs.base import AnyFSPath, FileSystem
from .db import ObjectDB
from .fs.base import AnyFSPath, FileSystem
from .hash_info import HashInfo
logger = logging.getLogger(__name__)
dvc/objects/fs/__init__.py
+127
View File
@@ -0,0 +1,127 @@
from urllib.parse import urlparse
from dvc.scheme import Schemes
from . import generic # noqa: F401
from .azure import AzureFileSystem
from .gdrive import GDriveFileSystem
from .gs import GSFileSystem
from .hdfs import HDFSFileSystem
from .http import HTTPFileSystem
from .https import HTTPSFileSystem
from .local import LocalFileSystem
from .oss import OSSFileSystem
from .s3 import S3FileSystem
from .ssh import SSHFileSystem
from .webdav import WebDAVFileSystem, WebDAVSFileSystem
from .webhdfs import WebHDFSFileSystem
FS_MAP = {
Schemes.AZURE: AzureFileSystem,
Schemes.GDRIVE: GDriveFileSystem,
Schemes.GS: GSFileSystem,
Schemes.HDFS: HDFSFileSystem,
Schemes.WEBHDFS: WebHDFSFileSystem,
Schemes.HTTP: HTTPFileSystem,
Schemes.HTTPS: HTTPSFileSystem,
Schemes.S3: S3FileSystem,
Schemes.SSH: SSHFileSystem,
Schemes.OSS: OSSFileSystem,
Schemes.WEBDAV: WebDAVFileSystem,
Schemes.WEBDAVS: WebDAVSFileSystem,
# NOTE: LocalFileSystem is the default
}
def get_fs_cls(remote_conf, scheme=None):
if not scheme:
scheme = urlparse(remote_conf["url"]).scheme
return FS_MAP.get(scheme, LocalFileSystem)
def get_fs_config(repo, config, **kwargs):
name = kwargs.get("name")
if name:
try:
remote_conf = config["remote"][name.lower()]
except KeyError:
from dvc.config import RemoteNotFoundError
raise RemoteNotFoundError(f"remote '{name}' doesn't exist")
else:
remote_conf = kwargs
return _resolve_remote_refs(repo, config, remote_conf)
def _resolve_remote_refs(repo, config, remote_conf):
# Support for cross referenced remotes.
# This will merge the settings, shadowing base ref with remote_conf.
# For example, having:
#
# dvc remote add server ssh://localhost
# dvc remote modify server user root
# dvc remote modify server ask_password true
#
# dvc remote add images remote://server/tmp/pictures
# dvc remote modify images user alice
# dvc remote modify images ask_password false
# dvc remote modify images password asdf1234
#
# Results on a config dictionary like:
#
# {
# "url": "ssh://localhost/tmp/pictures",
# "user": "alice",
# "password": "asdf1234",
# "ask_password": False,
# }
parsed = urlparse(remote_conf["url"])
if parsed.scheme != "remote":
return remote_conf
base = get_fs_config(repo, config, name=parsed.netloc)
cls, _, _ = get_cloud_fs(repo, **base)
relpath = parsed.path.lstrip("/").replace("/", cls.sep)
url = cls.sep.join((base["url"], relpath))
return {**base, **remote_conf, "url": url}
def get_cloud_fs(repo, **kwargs):
from dvc.config import ConfigError
from dvc.config_schema import SCHEMA, Invalid
repo_config = repo.config if repo else {}
core_config = repo_config.get("core", {})
remote_conf = get_fs_config(repo, 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 = core_config.get("jobs")
if jobs:
remote_conf["jobs"] = jobs
if "checksum_jobs" not in remote_conf:
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 and repo:
remote_conf["gdrive_credentials_tmp_dir"] = repo.tmp_dir
url = remote_conf.pop("url")
if issubclass(cls, WebDAVFileSystem):
# For WebDAVFileSystem, provided url is the base path itself, so it
# should be treated as being a root path.
fs_path = cls.root_marker
else:
fs_path = cls._strip_protocol(url) # pylint:disable=protected-access
extras = cls._get_kwargs_from_urls(url) # pylint:disable=protected-access
conf = {**extras, **remote_conf} # remote config takes priority
return cls, conf, fs_path
dvc/objects/fs/_callback.py
+235
View File
@@ -0,0 +1,235 @@
from contextlib import ExitStack
from functools import wraps
from typing import TYPE_CHECKING, Any, Dict, Optional, TypeVar, overload
import fsspec
from funcy import cached_property
if TYPE_CHECKING:
from typing import BinaryIO, Callable, TextIO, Union
from typing_extensions import ParamSpec
from dvc.progress import Tqdm
from dvc.ui._rich_progress import RichTransferProgress
_P = ParamSpec("_P")
_R = TypeVar("_R")
class FsspecCallback(fsspec.Callback):
"""FsspecCallback usable as a context manager, and a few helper methods."""
@overload
def wrap_attr(self, fobj: "BinaryIO", method: str = "read") -> "BinaryIO":
...
@overload
def wrap_attr(self, fobj: "TextIO", method: str = "read") -> "TextIO":
...
def wrap_attr(
self, fobj: "Union[TextIO, BinaryIO]", method: str = "read"
) -> "Union[TextIO, BinaryIO]":
from tqdm.utils import CallbackIOWrapper
wrapped = CallbackIOWrapper(self.relative_update, fobj, method)
return wrapped
def wrap_fn(self, fn: "Callable[_P, _R]") -> "Callable[_P, _R]":
@wraps(fn)
def wrapped(*args: "_P.args", **kwargs: "_P.kwargs") -> "_R":
res = fn(*args, **kwargs)
self.relative_update()
return res
return wrapped
def wrap_and_branch(self, fn: "Callable") -> "Callable":
"""
Wraps a function, and pass a new child callback to it.
When the function completes, we increment the parent callback by 1.
"""
wrapped = self.wrap_fn(fn)
@wraps(fn)
def func(path1: str, path2: str):
kw: Dict[str, Any] = {}
with self.branch(path1, path2, kw):
return wrapped(path1, path2, **kw)
return func
def __enter__(self):
return self
def __exit__(self, *exc_args):
self.close()
def close(self):
"""Handle here on exit."""
def relative_update(self, inc: int = 1) -> None:
inc = inc if inc is not None else 0
return super().relative_update(inc)
def absolute_update(self, value: int) -> None:
value = value if value is not None else self.value
return super().absolute_update(value)
@classmethod
def as_callback(
cls, maybe_callback: Optional["FsspecCallback"] = None
) -> "FsspecCallback":
if maybe_callback is None:
return DEFAULT_CALLBACK
return maybe_callback
@classmethod
def as_tqdm_callback(
cls,
callback: Optional["FsspecCallback"] = None,
**tqdm_kwargs: Any,
) -> "FsspecCallback":
return callback or TqdmCallback(**tqdm_kwargs)
@classmethod
def as_rich_callback(
cls, callback: Optional["FsspecCallback"] = None, **rich_kwargs
):
return callback or RichCallback(**rich_kwargs)
def branch(
self,
path_1: str,
path_2: str,
kwargs: Dict[str, Any],
child: "FsspecCallback" = None,
) -> "FsspecCallback":
child = kwargs["callback"] = child or DEFAULT_CALLBACK
return child
class NoOpCallback(FsspecCallback, fsspec.callbacks.NoOpCallback):
pass
class TqdmCallback(FsspecCallback):
def __init__(
self,
size: Optional[int] = None,
value: int = 0,
progress_bar: "Tqdm" = None,
**tqdm_kwargs,
):
tqdm_kwargs["total"] = size or -1
self._tqdm_kwargs = tqdm_kwargs
self._progress_bar = progress_bar
self._stack = ExitStack()
super().__init__(size=size, value=value)
@cached_property
def progress_bar(self):
from dvc.progress import Tqdm
progress_bar = (
self._progress_bar
if self._progress_bar is not None
else Tqdm(**self._tqdm_kwargs)
)
return self._stack.enter_context(progress_bar)
def __enter__(self):
return self
def close(self):
self._stack.close()
def set_size(self, size):
# Tqdm tries to be smart when to refresh,
# so we try to force it to re-render.
super().set_size(size)
self.progress_bar.refresh()
def call(self, hook_name=None, **kwargs):
self.progress_bar.update_to(self.value, total=self.size)
def branch(
self,
path_1: str,
path_2: str,
kwargs,
child: Optional[FsspecCallback] = None,
):
child = child or TqdmCallback(bytes=True, desc=path_1)
return super().branch(path_1, path_2, kwargs, child=child)
class RichCallback(FsspecCallback):
def __init__(
self,
size: Optional[int] = None,
value: int = 0,
progress: "RichTransferProgress" = None,
desc: str = None,
bytes: bool = False, # pylint: disable=redefined-builtin
unit: str = None,
disable: bool = False,
) -> None:
self._progress = progress
self.disable = disable
self._task_kwargs = {
"description": desc or "",
"bytes": bytes,
"unit": unit,
"total": size or 0,
"visible": False,
"progress_type": None if bytes else "summary",
}
self._stack = ExitStack()
super().__init__(size=size, value=value)
@cached_property
def progress(self):
from dvc.ui import ui
from dvc.ui._rich_progress import RichTransferProgress
if self._progress is not None:
return self._progress
progress = RichTransferProgress(
transient=True,
disable=self.disable,
console=ui.error_console,
)
return self._stack.enter_context(progress)
@cached_property
def task(self):
return self.progress.add_task(**self._task_kwargs)
def __enter__(self):
return self
def close(self):
self.progress.clear_task(self.task)
self._stack.close()
def call(self, hook_name=None, **kwargs):
self.progress.update(
self.task,
completed=self.value,
total=self.size,
visible=not self.disable,
)
def branch(
self, path_1, path_2, kwargs, child: Optional[FsspecCallback] = None
):
child = child or RichCallback(
progress=self.progress, desc=path_1, bytes=True
)
return super().branch(path_1, path_2, kwargs, child=child)
DEFAULT_CALLBACK = NoOpCallback()
dvc/fs/azure.py → dvc/objects/fs/azure.py
View File
dvc/fs/base.py → dvc/objects/fs/base.py
+3 -3
View File
@@ -131,9 +131,9 @@ class FileSystem:
return missing
def _check_requires(self, **kwargs):
from ..scheme import Schemes
from ..utils import format_link
from ..utils.pkg import PKG
from dvc.scheme import Schemes
from dvc.utils import format_link
from dvc.utils.pkg import PKG
missing = self.get_missing_deps()
if not missing:
dvc/fs/gdrive.py → dvc/objects/fs/gdrive.py
View File
dvc/fs/generic.py → dvc/objects/fs/generic.py
View File
dvc/fs/git.py → dvc/objects/fs/git.py
View File
dvc/fs/gs.py → dvc/objects/fs/gs.py
View File
dvc/fs/hdfs.py → dvc/objects/fs/hdfs.py
View File
dvc/fs/http.py → dvc/objects/fs/http.py
View File
dvc/fs/https.py → dvc/objects/fs/https.py
View File
dvc/fs/local.py → dvc/objects/fs/local.py
+1 -1
View File
@@ -6,9 +6,9 @@ import threading
from fsspec import AbstractFileSystem
from funcy import cached_property, wrap_prop
from dvc.fs import system
from dvc.scheme import Schemes
from . import system
from .base import FileSystem
from .utils import copyfile, makedirs, move, remove, tmp_fname
dvc/fs/memory.py → dvc/objects/fs/memory.py
View File
dvc/fs/oss.py → dvc/objects/fs/oss.py
View File
dvc/fs/path.py → dvc/objects/fs/path.py
View File
dvc/fs/s3.py → dvc/objects/fs/s3.py
View File
dvc/fs/ssh.py → dvc/objects/fs/ssh.py
View File
dvc/fs/system.py → dvc/objects/fs/system.py
View File
dvc/fs/utils.py → dvc/objects/fs/utils.py
+1 -8
View File
@@ -157,7 +157,7 @@ def copyfile(
try:
system.reflink(src, dest)
except OSError:
from dvc.fs._callback import FsspecCallback
from ._callback import FsspecCallback
with open(src, "rb") as fsrc, open(dest, "wb+") as fdest:
with FsspecCallback.as_tqdm_callback(
@@ -191,13 +191,6 @@ def tmp_fname(fname: "AnyFSPath" = "") -> "AnyFSPath":
return os.fspath(fname) + "." + uuid() + ".tmp"
def as_posix(path: str) -> str:
import ntpath
import posixpath
return path.replace(ntpath.sep, posixpath.sep)
@contextmanager
def as_atomic(
fs: "FileSystem", to_info: "AnyFSPath", create_parents: bool = False
dvc/fs/webdav.py → dvc/objects/fs/webdav.py
View File
dvc/fs/webhdfs.py → dvc/objects/fs/webhdfs.py
View File
dvc/objects/hash.py
+3 -5
View File
@@ -2,9 +2,8 @@ import hashlib
import logging
from typing import TYPE_CHECKING, Any, BinaryIO, Dict, Tuple
from dvc.fs._callback import DEFAULT_CALLBACK, FsspecCallback, TqdmCallback
from dvc.fs.utils import is_exec
from .fs._callback import DEFAULT_CALLBACK, FsspecCallback, TqdmCallback
from .fs.utils import is_exec
from .hash_info import HashInfo
from .istextfile import istextfile
from .meta import Meta
@@ -13,8 +12,7 @@ logger = logging.getLogger(__name__)
if TYPE_CHECKING:
from dvc.fs.base import AnyFSPath, FileSystem
from .fs.base import AnyFSPath, FileSystem
from .state import StateBase
dvc/objects/istextfile.py
+1 -1
View File
@@ -5,7 +5,7 @@
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from dvc.fs.base import AnyFSPath, FileSystem
from .fs.base import AnyFSPath, FileSystem
TEXT_CHARS = bytes(range(32, 127)) + b"\n\r\t\f\b"
DEFAULT_CHUNK_SIZE = 512
dvc/objects/state.py
+3 -4
View File
@@ -4,10 +4,9 @@ import logging
import os
from abc import ABC, abstractmethod
from dvc.fs.local import LocalFileSystem
from dvc.fs.system import inode as get_inode
from dvc.fs.utils import relpath
from .fs import LocalFileSystem
from .fs.system import inode as get_inode
from .fs.utils import relpath
from .hash_info import HashInfo
from .utils import get_mtime_and_size
dvc/objects/utils.py
+1 -1
View File
@@ -4,7 +4,7 @@ import json
from typing import TYPE_CHECKING, Tuple
if TYPE_CHECKING:
from dvc.fs.base import AnyFSPath, FileSystem
from .fs.base import AnyFSPath, FileSystem
def get_mtime_and_size(
dvc/output.py
+7 -5
View File
@@ -25,11 +25,13 @@ from .data.checkout import checkout
from .data.stage import stage as ostage
from .data.transfer import transfer as otransfer
from .data.tree import Tree
from .fs import get_cloud_fs
from .fs.base import RemoteMissingDepsError
from .fs.hdfs import HDFSFileSystem
from .fs.local import LocalFileSystem
from .fs.s3 import S3FileSystem
from .fs import (
HDFSFileSystem,
LocalFileSystem,
RemoteMissingDepsError,
S3FileSystem,
get_cloud_fs,
)
from .objects.errors import ObjectFormatError
from .scheme import Schemes
from .utils import relpath
dvc/repo/__init__.py
+5 -6
View File
@@ -15,7 +15,7 @@ from dvc.utils import env2bool
from dvc.utils.fs import path_isin
if TYPE_CHECKING:
from dvc.fs.base import FileSystem
from dvc.fs import FileSystem
from dvc.objects.file import HashFile
from dvc.repo.scm_context import SCMContext
from dvc.scm import Base
@@ -88,7 +88,7 @@ class Repo:
uninitialized: bool = False,
scm: "Base" = None,
):
from dvc.fs.local import localfs
from dvc.fs import localfs
from dvc.scm import SCM, SCMError
dvc_dir = None
@@ -157,8 +157,7 @@ class Repo:
from dvc.config import Config
from dvc.data.db import ODBManager
from dvc.data_cloud import DataCloud
from dvc.fs.git import GitFileSystem
from dvc.fs.local import localfs
from dvc.fs import GitFileSystem, localfs
from dvc.lock import LockNoop, make_lock
from dvc.objects.state import State, StateNoop
from dvc.repo.metrics import Metrics
@@ -280,7 +279,7 @@ class Repo:
return DvcIgnoreFilter(self.fs, self.root_dir)
def get_rev(self):
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
assert self.scm
if isinstance(self.fs, LocalFileSystem):
@@ -323,7 +322,7 @@ class Repo:
@classmethod
def find_root(cls, root=None, fs=None) -> str:
from dvc.fs.local import LocalFileSystem, localfs
from dvc.fs import LocalFileSystem, localfs
fs = fs or localfs
root = root or os.curdir
dvc/repo/brancher.py
+2 -2
View File
@@ -29,7 +29,7 @@ def brancher( # noqa: E302
yield ""
return
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
repo_root_parts = ()
if self.fs.path.isin(self.root_dir, self.scm.root_dir):
@@ -63,7 +63,7 @@ def brancher( # noqa: E302
)
try:
from dvc.fs.git import GitFileSystem
from dvc.fs import GitFileSystem
for sha, names in found_revs.items():
self.fs = GitFileSystem(scm=scm, rev=sha)
dvc/repo/diff.py
+1 -1
View File
@@ -112,7 +112,7 @@ def _paths_checksums(repo, targets):
def _output_paths(repo, targets):
from dvc.data.stage import stage as ostage
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
on_working_fs = isinstance(repo.fs, LocalFileSystem)
dvc/repo/experiments/executor/ssh.py
+1 -1
View File
@@ -7,7 +7,7 @@ from typing import TYPE_CHECKING, Callable, Iterable, Optional
from funcy import first
from dvc.fs.ssh import SSHFileSystem
from dvc.fs import SSHFileSystem
from dvc.repo.experiments.base import (
EXEC_BRANCH,
EXEC_CHECKPOINT,
dvc/repo/experiments/init.py
+1 -1
View File
@@ -162,7 +162,7 @@ def init_deps(stage: PipelineStage) -> List["Dependency"]:
from funcy import rpartial
from dvc.dependency import ParamsDependency
from dvc.fs.local import localfs
from dvc.fs import localfs
new_deps = [dep for dep in stage.deps if not dep.exists]
params, deps = lsplit(rpartial(isinstance, ParamsDependency), new_deps)
dvc/repo/graph.py
+1 -1
View File
@@ -1,6 +1,6 @@
from typing import TYPE_CHECKING, Iterator, List, Set
from dvc.fs.local import localfs
from dvc.fs import localfs
from dvc.utils.fs import path_isin
if TYPE_CHECKING:
dvc/repo/index.py
+2 -2
View File
@@ -21,7 +21,7 @@ if TYPE_CHECKING:
from dvc.data.tree import Tree
from dvc.dependency import Dependency, ParamsDependency
from dvc.fs.base import FileSystem
from dvc.fs import FileSystem
from dvc.objects.db import ObjectDB
from dvc.objects.hash_info import HashInfo
from dvc.output import Output
@@ -74,7 +74,7 @@ class Index:
return self.stage_collector.collect_repo(onerror=onerror)
def __repr__(self) -> str:
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
rev = "workspace"
if not isinstance(self.fs, LocalFileSystem):
dvc/repo/stage.py
+1 -1
View File
@@ -465,7 +465,7 @@ class StageLoad:
the collection.
"""
from dvc.dvcfile import is_valid_filename
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
scm = self.repo.scm
sep = self.fs.sep
dvc/rwlock.py
+1 -1
View File
@@ -6,7 +6,7 @@ from contextlib import contextmanager
from voluptuous import Invalid, Optional, Required, Schema
from .exceptions import DvcException
from .fs.local import localfs
from .fs import localfs
from .utils import relpath
INFO_SCHEMA = {Required("pid"): int, Required("cmd"): str}
dvc/utils/__init__.py
+1 -1
View File
@@ -317,7 +317,7 @@ def resolve_output(inp, out):
def resolve_paths(repo, out, always_local=False):
from urllib.parse import urlparse
from dvc.fs.local import localfs
from dvc.fs import localfs
from ..dvcfile import DVC_FILE_SUFFIX
from ..exceptions import DvcException
dvc/utils/serialize/_common.py
+1 -1
View File
@@ -9,7 +9,7 @@ from typing_extensions import Protocol
from dvc.exceptions import DvcException
if TYPE_CHECKING:
from dvc.fs.base import FileSystem
from dvc.fs import FileSystem
from dvc.types import AnyPath
dvc/utils/strictyaml.py
+1 -1
View File
@@ -25,7 +25,7 @@ if TYPE_CHECKING:
from ruamel.yaml import StreamMark
from voluptuous import MultipleInvalid
from dvc.fs.base import FileSystem
from dvc.fs import FileSystem
from dvc.ui import RichText
tests/func/experiments/executor/test_ssh.py
+1 -1
View File
@@ -6,7 +6,7 @@ from urllib.parse import urlparse
import pytest
from dvc_ssh.tests.cloud import TEST_SSH_KEY_PATH, TEST_SSH_USER
from dvc.fs.ssh import SSHFileSystem
from dvc.fs import SSHFileSystem
from dvc.repo.experiments.base import EXEC_HEAD, EXEC_MERGE
from dvc.repo.experiments.executor.base import ExecutorInfo, ExecutorResult
from dvc.repo.experiments.executor.ssh import SSHExecutor
tests/func/objects/db/test_index.py
+3 -3
View File
@@ -2,7 +2,7 @@ import pytest
from dvc.data.db import get_index
from dvc.exceptions import DownloadError, UploadError
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
from dvc.utils.fs import remove
from tests.utils import clean_staging
@@ -66,7 +66,7 @@ def test_clear_on_download_err(tmp_dir, dvc, index, mocker):
assert list(index.hashes())
mocker.patch("dvc.fs.generic.transfer", side_effect=Exception)
mocker.patch("dvc.objects.fs.generic.transfer", side_effect=Exception)
with pytest.raises(DownloadError):
dvc.pull()
assert not list(index.hashes())
@@ -83,7 +83,7 @@ def test_partial_upload(tmp_dir, dvc, index, mocker):
raise Exception("stop baz")
return original(self, from_file, to_info, name, **kwargs)
mocker.patch("dvc.fs.generic.transfer", unreliable_upload)
mocker.patch("dvc.objects.fs.generic.transfer", unreliable_upload)
with pytest.raises(UploadError):
dvc.push()
assert not list(index.hashes())
tests/func/test_add.py
+1 -2
View File
@@ -21,8 +21,7 @@ from dvc.exceptions import (
OverlappingOutputPathsError,
RecursiveAddingWhileUsingFilename,
)
from dvc.fs import system
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem, system
from dvc.objects.hash import file_md5
from dvc.objects.hash_info import HashInfo
from dvc.output import (
tests/func/test_checkout.py
+1 -2
View File
@@ -18,8 +18,7 @@ from dvc.exceptions import (
DvcException,
NoOutputOrStageError,
)
from dvc.fs import system
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem, system
from dvc.stage import Stage
from dvc.stage.exceptions import StageFileDoesNotExistError
from dvc.utils import relpath
tests/func/test_external_repo.py
+2 -2
View File
@@ -47,14 +47,14 @@ def test_source_change(erepo_dir):
def test_cache_reused(erepo_dir, mocker, local_cloud):
from dvc import fs
from dvc.objects.fs import generic
erepo_dir.add_remote(config=local_cloud.config)
with erepo_dir.chdir():
erepo_dir.dvc_gen("file", "text", commit="add file")
erepo_dir.dvc.push()
download_spy = mocker.spy(fs.generic, "transfer")
download_spy = mocker.spy(generic, "transfer")
# Use URL to prevent any fishy optimizations
url = f"file://{erepo_dir}"
tests/func/test_fs.py
+1 -2
View File
@@ -5,9 +5,8 @@ from os.path import join
import pytest
from dvc.fs import get_cloud_fs
from dvc.fs import LocalFileSystem, get_cloud_fs
from dvc.fs._callback import DEFAULT_CALLBACK, FsspecCallback
from dvc.fs.local import LocalFileSystem
from dvc.repo import Repo
tests/func/test_gc.py
+1 -1
View File
@@ -9,7 +9,7 @@ from git import Repo
from dvc.cli import main
from dvc.data.db.local import LocalObjectDB
from dvc.exceptions import CollectCacheError
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
from dvc.repo import Repo as DvcRepo
from dvc.utils.fs import remove
from tests.basic_env import TestDir, TestDvcGit
tests/func/test_ignore.py
+1 -1
View File
@@ -125,7 +125,7 @@ def test_ignore_collecting_dvcignores(tmp_dir, dvc, dname):
def test_ignore_on_branch(tmp_dir, scm, dvc):
from dvc.fs.git import GitFileSystem
from dvc.fs import GitFileSystem
tmp_dir.scm_gen({"foo": "foo", "bar": "bar"}, commit="add files")
tests/func/test_import.py
+1 -1
View File
@@ -308,7 +308,7 @@ def test_download_error_pulling_imported_stage(tmp_dir, dvc, erepo_dir):
remove(dst_cache)
with patch(
"dvc.fs.generic.transfer", side_effect=Exception
"dvc.objects.fs.generic.transfer", side_effect=Exception
), pytest.raises(DownloadError):
dvc.pull(["foo_imported.dvc"])
tests/func/test_odb.py
+2 -2
View File
@@ -180,7 +180,7 @@ def test_default_cache_type(dvc):
@pytest.mark.skipif(os.name == "nt", reason="Not supported for Windows.")
@pytest.mark.parametrize("group", [False, True])
def test_shared_cache(tmp_dir, dvc, group):
from dvc.fs.system import umask
from dvc.fs import system
if group:
with dvc.config.edit() as conf:
@@ -201,7 +201,7 @@ def test_shared_cache(tmp_dir, dvc, group):
actual[path] = oct(stat.S_IMODE(os.stat(path).st_mode))
file_mode = oct(0o444)
dir_mode = oct(0o2775 if group else (0o777 & ~umask))
dir_mode = oct(0o2775 if group else (0o777 & ~system.umask))
expected = {
os.path.join(cache_dir, "17"): dir_mode,
tests/func/test_remote.py
+7 -7
View File
@@ -169,14 +169,14 @@ def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc):
def test_partial_push_n_pull(tmp_dir, dvc, tmp_path_factory, local_remote):
from dvc import fs
from dvc.objects.fs import generic
foo = tmp_dir.dvc_gen({"foo": "foo content"})[0].outs[0]
bar = tmp_dir.dvc_gen({"bar": "bar content"})[0].outs[0]
baz = tmp_dir.dvc_gen({"baz": {"foo": "foo content"}})[0].outs[0]
# Faulty upload version, failing on foo
original = fs.generic.transfer
original = generic.transfer
odb = dvc.cloud.get_remote_odb("upstream")
def unreliable_upload(from_fs, from_info, to_fs, to_info, **kwargs):
@@ -186,7 +186,7 @@ def test_partial_push_n_pull(tmp_dir, dvc, tmp_path_factory, local_remote):
raise Exception("stop foo")
return original(from_fs, from_info, to_fs, to_info, **kwargs)
with patch.object(fs.generic, "transfer", unreliable_upload):
with patch.object(generic, "transfer", unreliable_upload):
with pytest.raises(UploadError) as upload_error_info:
dvc.push()
assert upload_error_info.value.amount == 2
@@ -200,7 +200,7 @@ def test_partial_push_n_pull(tmp_dir, dvc, tmp_path_factory, local_remote):
remove(dvc.odb.local.cache_dir)
baz._collect_used_dir_cache()
with patch.object(fs.generic, "transfer", side_effect=Exception):
with patch.object(generic, "transfer", side_effect=Exception):
with pytest.raises(DownloadError) as download_error_info:
dvc.pull()
# error count should be len(.dir + standalone file checksums)
@@ -214,7 +214,7 @@ def test_raise_on_too_many_open_files(
tmp_dir.dvc_gen({"file": "file content"})
mocker.patch(
"dvc.fs.generic.transfer",
"dvc.objects.fs.generic.transfer",
side_effect=OSError(errno.EMFILE, "Too many open files"),
)
@@ -263,12 +263,12 @@ def test_external_dir_resource_on_no_cache(tmp_dir, dvc, tmp_path_factory):
def test_push_order(tmp_dir, dvc, tmp_path_factory, mocker, local_remote):
from dvc import fs
from dvc.objects.fs import generic
foo = tmp_dir.dvc_gen({"foo": {"bar": "bar content"}})[0].outs[0]
tmp_dir.dvc_gen({"baz": "baz content"})
mocked_upload = mocker.spy(fs.generic, "transfer")
mocked_upload = mocker.spy(generic, "transfer")
dvc.push()
# foo .dir file should be uploaded after bar
tests/func/test_repro.py
+1 -2
View File
@@ -14,8 +14,7 @@ from dvc.exceptions import (
ReproductionError,
StagePathAsOutputError,
)
from dvc.fs import system
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem, system
from dvc.objects.hash import file_md5
from dvc.output import Output
from dvc.stage import Stage
tests/func/test_stage.py
+1 -1
View File
@@ -5,7 +5,7 @@ import pytest
from dvc.cli import main
from dvc.dvcfile import SingleStageFile
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
from dvc.output import Output
from dvc.repo import Repo, lock_repo
from dvc.stage import PipelineStage, Stage
tests/unit/fs/test_azure.py
+1 -1
View File
@@ -1,6 +1,6 @@
import pytest
from dvc.fs.azure import AzureAuthError, AzureFileSystem
from dvc.fs import AzureAuthError, AzureFileSystem
container_name = "container-name"
connection_string = (
tests/unit/fs/test_base.py
+1 -1
View File
@@ -1,6 +1,6 @@
import pytest
from dvc.fs.base import FileSystem, RemoteMissingDepsError
from dvc.fs import FileSystem, RemoteMissingDepsError
@pytest.mark.parametrize(
tests/unit/fs/test_fs.py
+10 -7
View File
@@ -1,13 +1,16 @@
import pytest
from dvc.config import RemoteNotFoundError
from dvc.fs import get_fs_cls, get_fs_config
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
from dvc.fs import (
HDFSFileSystem,
HTTPFileSystem,
HTTPSFileSystem,
LocalFileSystem,
S3FileSystem,
SSHFileSystem,
get_fs_cls,
get_fs_config,
)
@pytest.mark.parametrize(
tests/unit/fs/test_path.py
+1 -1
View File
@@ -1,6 +1,6 @@
import pytest
from dvc.fs.path import Path
from dvc.fs import Path
@pytest.mark.parametrize("prefix", ["", "/"])
tests/unit/fs/test_s3.py
+1 -1
View File
@@ -3,7 +3,7 @@ import os
import pytest
from dvc.config import ConfigError
from dvc.fs.s3 import S3FileSystem
from dvc.fs import S3FileSystem
bucket_name = "bucket-name"
prefix = "some/prefix"
tests/unit/fs/test_ssh.py
+3 -3
View File
@@ -6,7 +6,7 @@ import os
import pytest
from mock import mock_open, patch
from dvc.fs.ssh import DEFAULT_PORT, SSHFileSystem
from dvc.fs import DEFAULT_SSH_PORT, SSHFileSystem
def test_get_kwargs_from_urls():
@@ -44,7 +44,7 @@ def test_init():
def test_ssh_ask_password(mocker):
mocker.patch("dvc.fs.ssh.ask_password", return_value="fish")
mocker.patch("dvc.objects.fs.ssh.ask_password", return_value="fish")
fs = SSHFileSystem(user="test", host="2.2.2.2", ask_password=True)
assert fs.fs_args["password"] == fs.fs_args["passphrase"] == "fish"
@@ -99,7 +99,7 @@ def test_ssh_user(mock_file, config, expected_user):
[
({"host": "example.com"}, 1234),
({"host": "example.com", "port": 4321}, 4321),
({"host": "not_in_ssh_config.com"}, DEFAULT_PORT),
({"host": "not_in_ssh_config.com"}, DEFAULT_SSH_PORT),
({"host": "not_in_ssh_config.com", "port": 2222}, 2222),
],
)
tests/unit/objects/db/test_local.py
+1 -1
View File
@@ -4,7 +4,7 @@ import os
import pytest
from dvc.data.db.local import LocalObjectDB
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
from dvc.objects.hash_info import HashInfo
tests/unit/output/test_load.py
+1 -2
View File
@@ -3,8 +3,7 @@
import pytest
from dvc import output
from dvc.fs.local import LocalFileSystem
from dvc.fs.s3 import S3FileSystem
from dvc.fs import LocalFileSystem, S3FileSystem
from dvc.output import Output
from dvc.stage import Stage
tests/unit/output/test_output.py
+1 -1
View File
@@ -5,7 +5,7 @@ import pytest
from funcy import first
from voluptuous import MultipleInvalid, Schema
from dvc.fs.base import RemoteMissingDepsError
from dvc.fs import RemoteMissingDepsError
from dvc.ignore import _no_match
from dvc.output import CHECKSUM_SCHEMA, Output
from dvc.stage import Stage
tests/unit/remote/test_base.py
+1 -1
View File
@@ -2,7 +2,7 @@ import math
import posixpath
from unittest import mock
from dvc.fs.base import FileSystem
from dvc.fs import FileSystem
from dvc.objects.db import ObjectDB
tests/unit/remote/test_gdrive.py
+1 -1
View File
@@ -1,6 +1,6 @@
import pytest
from dvc.fs.gdrive import GDriveAuthError, GDriveFileSystem
from dvc.fs import GDriveAuthError, GDriveFileSystem
USER_CREDS_TOKEN_REFRESH_ERROR = '{"access_token": "", "client_id": "", "client_secret": "", "refresh_token": "", "token_expiry": "", "token_uri": "https://oauth2.googleapis.com/token", "user_agent": null, "revoke_uri": "https://oauth2.googleapis.com/revoke", "id_token": null, "id_token_jwt": null, "token_response": {"access_token": "", "expires_in": 3600, "scope": "https://www.googleapis.com/auth/drive.appdata https://www.googleapis.com/auth/drive", "token_type": "Bearer"}, "scopes": ["https://www.googleapis.com/auth/drive", "https://www.googleapis.com/auth/drive.appdata"], "token_info_uri": "https://oauth2.googleapis.com/tokeninfo", "invalid": true, "_class": "OAuth2Credentials", "_module": "oauth2client.client"}' # noqa: E501
tests/unit/remote/test_http.py
+1 -1
View File
@@ -2,7 +2,7 @@ import ssl
from mock import patch
from dvc.fs.http import HTTPFileSystem
from dvc.fs import HTTPFileSystem
def test_public_auth_method(dvc):
tests/unit/remote/test_oss.py
+1 -1
View File
@@ -1,4 +1,4 @@
from dvc.fs.oss import OSSFileSystem
from dvc.fs import OSSFileSystem
bucket_name = "bucket-name"
endpoint = "endpoint"
tests/unit/remote/test_remote.py
+1 -3
View File
@@ -3,9 +3,7 @@ import os
import pytest
from dvc.data.db import get_index
from dvc.fs import get_cloud_fs
from dvc.fs.gs import GSFileSystem
from dvc.fs.s3 import S3FileSystem
from dvc.fs import GSFileSystem, S3FileSystem, get_cloud_fs
def test_remote_with_hash_jobs(dvc):
tests/unit/remote/test_webdav.py
+2 -3
View File
@@ -2,8 +2,7 @@ from unittest.mock import patch
import pytest
from dvc.fs import get_cloud_fs
from dvc.fs.webdav import WebDAVFileSystem, WebDAVSFileSystem
from dvc.fs import WebDAVFileSystem, WebDAVSFileSystem, get_cloud_fs
from tests.utils.asserts import issubset
url_fmt = "{scheme}://{user}@example.com/public.php/webdav"
@@ -66,7 +65,7 @@ def test_token():
)
@patch("dvc.fs.webdav.ask_password")
@patch("dvc.objects.fs.webdav.ask_password")
def test_ask_password(ask_password_mocked):
ask_password_mocked.return_value = "pass"
host = "host"
tests/unit/remote/test_webhdfs.py
+1 -1
View File
@@ -3,7 +3,7 @@ from unittest.mock import Mock, create_autospec
import pytest
import requests
from dvc.fs.webhdfs import WebHDFSFileSystem
from dvc.fs import WebHDFSFileSystem
host = "host"
kerberos = False
tests/unit/stage/test_utils.py
+1 -1
View File
@@ -1,6 +1,6 @@
import os
from dvc.fs.local import localfs
from dvc.fs import localfs
from dvc.stage.utils import resolve_paths
tests/unit/test_config.py
+1 -1
View File
@@ -4,7 +4,7 @@ import textwrap
import pytest
from dvc.config import Config
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
@pytest.mark.parametrize(
tests/unit/test_context.py
+1 -1
View File
@@ -3,7 +3,7 @@ from math import pi
import pytest
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
from dvc.parsing import DEFAULT_PARAMS_FILE
from dvc.parsing.context import (
Context,
tests/unit/test_ignore.py
+1 -1
View File
@@ -7,7 +7,7 @@ from dvc.ignore import DvcIgnorePatterns
def mock_dvcignore(dvcignore_path, patterns):
from dvc.fs.local import localfs
from dvc.fs import localfs
fs = MagicMock()
fs.path = localfs.path
tests/unit/utils/test_fs.py
+1 -2
View File
@@ -4,8 +4,7 @@ import os
import pytest
import dvc
from dvc.fs import system
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem, system
from dvc.objects.utils import get_mtime_and_size
from dvc.utils import relpath
from dvc.utils.fs import (
tests/unit/utils/test_stream.py
+1 -1
View File
@@ -1,7 +1,7 @@
import pytest
from dvc.data.stream import HashedStreamReader
from dvc.fs.local import LocalFileSystem
from dvc.fs import LocalFileSystem
from dvc.objects.hash import file_md5
from dvc.objects.istextfile import DEFAULT_CHUNK_SIZE, istextfile
tests/utils/__init__.py
+1 -1
View File
@@ -46,7 +46,7 @@ def dump_sv(stream, metrics, delimiter=",", header=True):
def clean_staging():
from dvc.data.stage import _STAGING_MEMFS_PATH
from dvc.fs.memory import MemoryFileSystem
from dvc.fs import MemoryFileSystem
try:
MemoryFileSystem().fs.rm(