diff --git a/dvc/data/db/reference.py b/dvc/data/db/reference.py index b9fde6be0..01589f49c 100644 --- a/dvc/data/db/reference.py +++ b/dvc/data/db/reference.py @@ -10,9 +10,8 @@ from dvc.scheme import Schemes from ..reference import ReferenceHashFile if TYPE_CHECKING: - from dvc.fs.base import FileSystem + from dvc.fs.base import AnyFSPath, FileSystem from dvc.hash_info import HashInfo - from dvc.types import AnyPath logger = logging.getLogger(__name__) @@ -55,8 +54,8 @@ class ReferenceObjectDB(ObjectDB): def _add_file( self, from_fs: "FileSystem", - from_info: "AnyPath", - to_info: "AnyPath", + from_info: "AnyFSPath", + to_info: "AnyFSPath", hash_info: "HashInfo", hardlink: bool = False, ): diff --git a/dvc/data/reference.py b/dvc/data/reference.py index ec391bf68..f0f257882 100644 --- a/dvc/data/reference.py +++ b/dvc/data/reference.py @@ -8,10 +8,9 @@ from dvc.objects.errors import ObjectFormatError from dvc.objects.file import HashFile if TYPE_CHECKING: - from dvc.fs.base import FileSystem + from dvc.fs.base import AnyFSPath, FileSystem from dvc.hash_info import HashInfo from dvc.objects.db import ObjectDB - from dvc.types import AnyPath logger = logging.getLogger(__name__) @@ -24,7 +23,7 @@ class ReferenceHashFile(HashFile): def __init__( self, - fs_path: "AnyPath", + fs_path: "AnyFSPath", fs: "FileSystem", hash_info: "HashInfo", checksum: Optional[str] = None, diff --git a/dvc/data/stage.py b/dvc/data/stage.py index 4ab026dd5..591530360 100644 --- a/dvc/data/stage.py +++ b/dvc/data/stage.py @@ -16,9 +16,8 @@ from .db.reference import ReferenceObjectDB from .meta import Meta if TYPE_CHECKING: - from dvc.fs.base import FileSystem + from dvc.fs.base import AnyFSPath, FileSystem from dvc.objects.db import ObjectDB - from dvc.types import AnyPath logger = logging.getLogger(__name__) @@ -287,7 +286,7 @@ def _stage_external_tree_info(odb, tree, name): def stage( odb: "ObjectDB", - fs_path: "AnyPath", + fs_path: "AnyFSPath", fs: "FileSystem", name: str, upload: bool = False, diff --git a/dvc/fs/azure.py b/dvc/fs/azure.py index f9eeb0640..783ebad69 100644 --- a/dvc/fs/azure.py +++ b/dvc/fs/azure.py @@ -10,7 +10,7 @@ from dvc.exceptions import DvcException from dvc.scheme import Schemes from dvc.utils import format_link -from .fsspec_wrapper import ObjectFSWrapper +from .base import ObjectFileSystem logger = logging.getLogger(__name__) _DEFAULT_CREDS_STEPS = ( @@ -38,7 +38,7 @@ def _az_config(): # pylint:disable=abstract-method -class AzureFileSystem(ObjectFSWrapper): +class AzureFileSystem(ObjectFileSystem): scheme = Schemes.AZURE PARAM_CHECKSUM = "etag" REQUIRES = { diff --git a/dvc/fs/base.py b/dvc/fs/base.py index 7826cbcf7..cd82a1b99 100644 --- a/dvc/fs/base.py +++ b/dvc/fs/base.py @@ -1,10 +1,22 @@ import contextlib import logging import os +import shutil from concurrent.futures import ThreadPoolExecutor, as_completed from functools import partialmethod from multiprocessing import cpu_count -from typing import ClassVar, Dict, Optional +from typing import ( + IO, + TYPE_CHECKING, + Any, + ClassVar, + Dict, + Iterator, + List, + Optional, + Union, + overload, +) from funcy import cached_property from tqdm.utils import CallbackIOWrapper @@ -15,9 +27,22 @@ from dvc.ui import ui from dvc.utils import tmp_fname from dvc.utils.fs import makedirs, move +if TYPE_CHECKING: + from fsspec.spec import AbstractFileSystem + from typing_extensions import Literal + + logger = logging.getLogger(__name__) +FSPath = str +AnyFSPath = str + +# An info() entry, might evolve to a TypedDict +# in the future (e.g for properly type 'size' etc). +Entry = Dict[str, Any] + + class RemoteActionNotImplemented(DvcException): def __init__(self, action, scheme): m = f"{action} is not supported for {scheme} remotes" @@ -38,7 +63,7 @@ class FileSystem: HASH_JOBS = max(1, min(4, cpu_count() // 2)) LIST_OBJECT_PAGE_SIZE = 1000 TRAVERSE_WEIGHT_MULTIPLIER = 5 - TRAVERSE_PREFIX_LEN = 3 + TRAVERSE_PREFIX_LEN = 2 TRAVERSE_THRESHOLD_SIZE = 500000 CAN_TRAVERSE = True @@ -53,9 +78,11 @@ class FileSystem: self.jobs = kwargs.get("jobs") or self._JOBS self.hash_jobs = kwargs.get("checksum_jobs") or self.HASH_JOBS self._config = kwargs + self.fs_args = {"skip_instance_cache": True} + self.fs_args.update(self._prepare_credentials(**kwargs)) @property - def config(self): + def config(self) -> Dict[str, Any]: return self._config @cached_property @@ -65,21 +92,37 @@ class FileSystem: return Path(self.sep) @classmethod - def _strip_protocol(cls, path: str): + def _strip_protocol(cls, path: str) -> str: return path - def unstrip_protocol(self, path): + def unstrip_protocol(self, path: str) -> str: return path + @cached_property + def fs(self) -> "AbstractFileSystem": + raise NotImplementedError + @staticmethod - def _get_kwargs_from_urls(urlpath): # pylint:disable=unused-argument + def _get_kwargs_from_urls(urlpath: str) -> "Dict[str, Any]": + from fsspec.utils import infer_storage_options + + options = infer_storage_options(urlpath) + options.pop("path", None) + options.pop("protocol", None) + return options + + def _prepare_credentials( + self, **config: Dict[str, Any] # pylint: disable=unused-argument + ) -> Dict[str, Any]: + """Prepare the arguments for authentication to the + host filesystem""" return {} @classmethod - def get_missing_deps(cls): + def get_missing_deps(cls) -> List[str]: import importlib - missing = [] + missing: List[str] = [] for package, module in cls.REQUIRES.items(): try: importlib.import_module(module) @@ -127,90 +170,155 @@ class FileSystem: f"dependencies: {missing}. {hint}" ) - def checksum(self, path) -> str: - raise NotImplementedError + def isdir(self, path: AnyFSPath) -> bool: + return self.fs.isdir(path) - def open(self, path, mode: str = "r", encoding: str = None, **kwargs): - raise RemoteActionNotImplemented("open", self.scheme) + def isfile(self, path: AnyFSPath) -> bool: + return self.fs.isfile(path) - def exists(self, path) -> bool: - raise NotImplementedError + def is_empty(self, path: AnyFSPath) -> bool: + entry = self.info(path) + if entry["type"] == "directory": + return not self.fs.ls(path) + return entry["size"] == 0 - # pylint: disable=unused-argument + def open( + self, + path: AnyFSPath, + mode: str = "r", + encoding: Optional[str] = None, + **kwargs, + ) -> "IO": # pylint: disable=arguments-differ + return self.fs.open(path, mode=mode, encoding=encoding, **kwargs) - def isdir(self, path): - """Optional: Overwrite only if the remote has a way to distinguish - between a directory and a file. - """ - return False + def checksum(self, path: AnyFSPath) -> str: + return self.fs.checksum(path) - def isfile(self, path): - """Optional: Overwrite only if the remote has a way to distinguish - between a directory and a file. - """ - return True + def copy(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: + self.makedirs(self.path.parent(to_info)) + self.fs.copy(from_info, to_info) - def iscopy(self, path): - """Check if this file is an independent copy.""" - return False # We can't be sure by default + def exists(self, path: AnyFSPath) -> bool: + return self.fs.exists(path) - def walk(self, top, topdown=True, **kwargs): - """Return a generator with (root, dirs, files).""" - raise NotImplementedError + def lexists(self, path: AnyFSPath) -> bool: + return self.fs.lexists(path) - def find(self, path, prefix=None): - """Return a generator with `str`s to all the files. + def symlink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: + try: + return self.fs.symlink(from_info, to_info) + except AttributeError: + raise RemoteActionNotImplemented("symlink", self.scheme) - Optional kwargs: - prefix (bool): If true `path` will be treated as a prefix - rather than directory path. - """ - raise NotImplementedError + def hardlink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: + try: + return self.fs.hardlink(from_info, to_info) + except AttributeError: + raise RemoteActionNotImplemented("hardlink", self.scheme) - def is_empty(self, path): - return False + def reflink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: + try: + return self.fs.reflink(from_info, to_info) + except AttributeError: + raise RemoteActionNotImplemented("reflink", self.scheme) - def info(self, path): - raise NotImplementedError + def is_symlink(self, path: AnyFSPath) -> bool: + try: + return self.fs.is_symlink(path) + except AttributeError: + return False - def getsize(self, path): + def is_hardlink(self, path: AnyFSPath) -> bool: + try: + return self.fs.is_hardlink(path) + except AttributeError: + return False + + def iscopy(self, path: AnyFSPath) -> bool: + return self.is_symlink(path) or self.is_hardlink(path) + + @overload + def ls( + self, path: AnyFSPath, detail: "Literal[True]" + ) -> "Iterator[Entry]": + ... + + @overload + def ls(self, path: AnyFSPath, detail: "Literal[False]") -> Iterator[str]: + ... + + def ls(self, path, detail=False, **kwargs): + return self.fs.ls(path, detail=detail) + + def find( + self, + path: AnyFSPath, + prefix: bool = False, # pylint: disable=unused-argument + ) -> Iterator[str]: + yield from self.fs.find(path) + + def move(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: + self.fs.move(from_info, to_info) + + def remove(self, path: AnyFSPath) -> None: + self.fs.rm_file(path) + + def info(self, path: AnyFSPath) -> "Entry": + return self.fs.info(path) + + def makedirs(self, path: AnyFSPath, **kwargs: Any) -> None: + self.fs.makedirs(path, exist_ok=kwargs.pop("exist_ok", True)) + + def put_file( + self, + from_file: AnyFSPath, + to_info: AnyFSPath, + callback: Any = DEFAULT_CALLBACK, + **kwargs, + ) -> None: + self.fs.put_file(from_file, to_info, callback=callback, **kwargs) + self.fs.invalidate_cache(self.path.parent(to_info)) + + def get_file( + self, + from_info: AnyFSPath, + to_info: AnyFSPath, + callback: Any = DEFAULT_CALLBACK, + **kwargs, + ) -> None: + self.fs.get_file(from_info, to_info, callback=callback, **kwargs) + + def upload_fobj(self, fobj: IO, to_info: AnyFSPath, **kwargs) -> None: + self.makedirs(self.path.parent(to_info)) + with self.open(to_info, "wb") as fdest: + shutil.copyfileobj( + fobj, + fdest, + length=getattr(fdest, "blocksize", None), # type: ignore + ) + + def walk( + self, + path: AnyFSPath, + topdown: bool = True, + **kwargs: Any, + ): + return self.fs.walk(path, topdown=topdown, **kwargs) + + def getsize(self, path: AnyFSPath) -> Optional[int]: return self.info(path).get("size") - def remove(self, path): - raise RemoteActionNotImplemented("remove", self.scheme) - - def makedirs(self, path, **kwargs): - """Optional: Implement only if the remote needs to create - directories before copying/linking/moving data - """ - - def move(self, from_info, to_info): - self.copy(from_info, to_info) - self.remove(from_info) - - def copy(self, from_info, to_info): - raise RemoteActionNotImplemented("copy", self.scheme) - - def symlink(self, from_info, to_info): - raise RemoteActionNotImplemented("symlink", self.scheme) - - def hardlink(self, from_info, to_info): - raise RemoteActionNotImplemented("hardlink", self.scheme) - - def reflink(self, from_info, to_info): - raise RemoteActionNotImplemented("reflink", self.scheme) - # pylint: enable=unused-argument def upload( self, - from_info, - to_info, - total=None, - desc=None, + from_info: Union[AnyFSPath, IO], + to_info: AnyFSPath, + total: int = None, + desc: str = None, callback=None, - no_progress_bar=False, - **pbar_args, + no_progress_bar: bool = False, + **pbar_args: Any, ): is_file_obj = hasattr(from_info, "read") method = "upload_fobj" if is_file_obj else "put_file" @@ -246,6 +354,7 @@ class FileSystem: # pylint: disable=no-member return self.upload_fobj(wrapped, to_info, size=total) + assert isinstance(from_info, str) logger.debug("Uploading '%s' to '%s'", from_info, to_info) # pylint: disable=no-member return self.put_file( @@ -254,14 +363,14 @@ class FileSystem: def download( self, - from_info, - to_info, - name=None, + from_info: AnyFSPath, + to_info: AnyFSPath, + name: str = None, callback=None, - no_progress_bar=False, - jobs=None, - _only_file=False, - **kwargs, + no_progress_bar: bool = False, + jobs: int = None, + _only_file: bool = False, + **kwargs: Any, ): from .local import localfs @@ -295,10 +404,10 @@ class FileSystem: def _download_dir( self, - from_info, - to_info, + from_info: AnyFSPath, + to_info: AnyFSPath, callback=DEFAULT_CALLBACK, - jobs=None, + jobs: int = None, **kwargs, ): from .local import localfs @@ -339,10 +448,10 @@ class FileSystem: def _download_file( self, - from_info, - to_info, + from_info: AnyFSPath, + to_info: AnyFSPath, callback=DEFAULT_CALLBACK, - ): + ) -> None: from .local import localfs makedirs(localfs.path.parent(to_info), exist_ok=True) @@ -360,3 +469,59 @@ class FileSystem: raise move(tmp_file, to_info) + + +class ObjectFileSystem(FileSystem): # pylint: disable=abstract-method + TRAVERSE_PREFIX_LEN = 3 + + def makedirs(self, path: AnyFSPath, **kwargs: Any) -> None: + # For object storages make this method a no-op. The original + # fs.makedirs() method will only check if the bucket exists + # and create if it doesn't though we don't want to support + # that behavior, and the check will cost some time so we'll + # simply ignore all mkdir()/makedirs() calls. + return None + + def _isdir(self, path: AnyFSPath) -> bool: + # Directory in object storages are interpreted differently + # among different fsspec providers, so this logic is a temporary + # measure for us to adapt as of now. It checks whether it is a + # directory (as in a prefix with contents) or whether it is an empty + # file where it's name ends with a forward slash + + entry = self.info(path) + return entry["type"] == "directory" or ( + entry["size"] == 0 + and entry["type"] == "file" + and entry["name"].endswith("/") + ) + + def isdir(self, path: AnyFSPath) -> bool: + try: + return self._isdir(path) + except FileNotFoundError: + return False + + def isfile(self, path: AnyFSPath) -> bool: + try: + return not self._isdir(path) + except FileNotFoundError: + return False + + def find(self, path: AnyFSPath, prefix: bool = False) -> Iterator[str]: + if prefix: + with_prefix = self.path.parent(path) + files = self.fs.find(with_prefix, prefix=self.path.parts(path)[-1]) + else: + with_prefix = path + files = self.fs.find(path) + + # When calling find() on a file, it returns the same file in a list. + # For object-based storages, the same behavior applies to empty + # directories since they are represented as files. This condition + # checks whether we should yield an empty list (if it is an empty + # directory) or just yield the file itself. + if len(files) == 1 and files[0] == with_prefix and self.isdir(path): + return None + + yield from files diff --git a/dvc/fs/dvc.py b/dvc/fs/dvc.py index be76ed6d5..e7218e3ec 100644 --- a/dvc/fs/dvc.py +++ b/dvc/fs/dvc.py @@ -7,7 +7,7 @@ from fsspec import AbstractFileSystem from funcy import cached_property, wrap_prop from ._callback import DEFAULT_CALLBACK -from .fsspec_wrapper import FSSpecWrapper +from .base import FileSystem if typing.TYPE_CHECKING: from dvc.types import AnyPath @@ -167,7 +167,7 @@ class _DvcFileSystem(AbstractFileSystem): # pylint:disable=abstract-method raise NotImplementedError -class DvcFileSystem(FSSpecWrapper): +class DvcFileSystem(FileSystem): scheme = "local" PARAM_CHECKSUM = "md5" diff --git a/dvc/fs/fsspec_wrapper.py b/dvc/fs/fsspec_wrapper.py deleted file mode 100644 index 2adcc4d09..000000000 --- a/dvc/fs/fsspec_wrapper.py +++ /dev/null @@ -1,236 +0,0 @@ -import shutil -from typing import IO, TYPE_CHECKING, Any, Dict, Iterator, Optional, overload - -from funcy import cached_property - -from ._callback import DEFAULT_CALLBACK -from .base import FileSystem, RemoteActionNotImplemented - -FSPath = str -AnyFSPath = str - -if TYPE_CHECKING: - from typing_extensions import Literal - - -# An info() entry, might evolve to a TypedDict -# in the future (e.g for properly type 'size' etc). -Entry = Dict[str, Any] - - -# pylint: disable=no-member -class FSSpecWrapper(FileSystem): - TRAVERSE_PREFIX_LEN = 2 - - def __init__(self, **kwargs): - super().__init__(**kwargs) - self.fs_args = {"skip_instance_cache": True} - self.fs_args.update(self._prepare_credentials(**kwargs)) - - @staticmethod - def _get_kwargs_from_urls(urlpath: str) -> "Dict[str, Any]": - from fsspec.utils import infer_storage_options - - options = infer_storage_options(urlpath) - options.pop("path", None) - options.pop("protocol", None) - return options - - @cached_property - def fs(self): - raise NotImplementedError - - def _prepare_credentials( - self, **config: Dict[str, Any] # pylint: disable=unused-argument - ) -> Dict[str, Any]: - """Prepare the arguments for authentication to the - host filesystem""" - return {} - - def isdir(self, path: AnyFSPath) -> bool: - return self.fs.isdir(path) - - def isfile(self, path: AnyFSPath) -> bool: - return self.fs.isfile(path) - - def is_empty(self, path: AnyFSPath) -> bool: - entry = self.info(path) - if entry["type"] == "directory": - return not self.fs.ls(path) - return entry["size"] == 0 - - def open( - self, - path: AnyFSPath, - mode: str = "r", - encoding: Optional[str] = None, - **kwargs, - ) -> "IO": # pylint: disable=arguments-differ - return self.fs.open(path, mode=mode, encoding=encoding, **kwargs) - - def checksum(self, path: AnyFSPath) -> str: - return self.fs.checksum(path) - - def copy(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: - self.makedirs(self.path.parent(to_info)) - self.fs.copy(from_info, to_info) - - def exists(self, path: AnyFSPath) -> bool: - return self.fs.exists(path) - - def lexists(self, path: AnyFSPath) -> bool: - return self.fs.lexists(path) - - def symlink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: - try: - return self.fs.symlink(from_info, to_info) - except AttributeError: - raise RemoteActionNotImplemented("symlink", self.scheme) - - def hardlink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: - try: - return self.fs.hardlink(from_info, to_info) - except AttributeError: - raise RemoteActionNotImplemented("hardlink", self.scheme) - - def reflink(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: - try: - return self.fs.reflink(from_info, to_info) - except AttributeError: - raise RemoteActionNotImplemented("reflink", self.scheme) - - def is_symlink(self, path: AnyFSPath) -> bool: - try: - return self.fs.is_symlink(path) - except AttributeError: - return False - - def is_hardlink(self, path: AnyFSPath) -> bool: - try: - return self.fs.is_hardlink(path) - except AttributeError: - return False - - def iscopy(self, path: AnyFSPath) -> bool: - return self.is_symlink(path) or self.is_hardlink(path) - - @overload - def ls( - self, path: AnyFSPath, detail: "Literal[True]" - ) -> "Iterator[Entry]": - ... - - @overload - def ls(self, path: AnyFSPath, detail: "Literal[False]") -> Iterator[str]: - ... - - def ls(self, path, detail=False, **kwargs): - return self.fs.ls(path, detail=detail) - - def find(self, path, prefix=None): - yield from self.fs.find(path) - - def move(self, from_info: AnyFSPath, to_info: AnyFSPath) -> None: - self.fs.move(from_info, to_info) - - def remove(self, path: AnyFSPath) -> None: - self.fs.rm_file(path) - - def info(self, path: AnyFSPath) -> "Entry": - return self.fs.info(path) - - def makedirs(self, path: AnyFSPath, **kwargs) -> None: - self.fs.makedirs(path, exist_ok=kwargs.pop("exist_ok", True)) - - def put_file( - self, - from_file: AnyFSPath, - to_info: AnyFSPath, - callback: Any = DEFAULT_CALLBACK, - **kwargs, - ) -> None: - self.fs.put_file(from_file, to_info, callback=callback, **kwargs) - self.fs.invalidate_cache(self.path.parent(to_info)) - - def get_file( - self, - from_info: AnyFSPath, - to_info: AnyFSPath, - callback: Any = DEFAULT_CALLBACK, - **kwargs, - ) -> None: - self.fs.get_file(from_info, to_info, callback=callback, **kwargs) - - def upload_fobj(self, fobj: IO, to_info: AnyFSPath, **kwargs) -> None: - self.makedirs(self.path.parent(to_info)) - with self.open(to_info, "wb") as fdest: - shutil.copyfileobj( - fobj, - fdest, - length=getattr(fdest, "blocksize", None), # type: ignore - ) - - def walk( - self, - top: AnyFSPath, - topdown: bool = True, - **kwargs: Any, - ): - return self.fs.walk(top, topdown=topdown, **kwargs) - - -# pylint: disable=abstract-method -class ObjectFSWrapper(FSSpecWrapper): - TRAVERSE_PREFIX_LEN = 3 - - def makedirs(self, path: AnyFSPath, **kwargs) -> None: - # For object storages make this method a no-op. The original - # fs.makedirs() method will only check if the bucket exists - # and create if it doesn't though we don't want to support - # that behavior, and the check will cost some time so we'll - # simply ignore all mkdir()/makedirs() calls. - return None - - def _isdir(self, path: AnyFSPath) -> bool: - # Directory in object storages are interpreted differently - # among different fsspec providers, so this logic is a temporary - # measure for us to adapt as of now. It checks whether it is a - # directory (as in a prefix with contents) or whether it is an empty - # file where it's name ends with a forward slash - - entry = self.info(path) - return entry["type"] == "directory" or ( - entry["size"] == 0 - and entry["type"] == "file" - and entry["name"].endswith("/") - ) - - def isdir(self, path: AnyFSPath) -> bool: - try: - return self._isdir(path) - except FileNotFoundError: - return False - - def isfile(self, path: AnyFSPath) -> bool: - try: - return not self._isdir(path) - except FileNotFoundError: - return False - - def find(self, path, prefix=None): - if prefix: - with_prefix = self.path.parent(path) - files = self.fs.find(with_prefix, prefix=self.path.parts(path)[-1]) - else: - with_prefix = path - files = self.fs.find(path) - - # When calling find() on a file, it returns the same file in a list. - # For object-based storages, the same behavior applies to empty - # directories since they are represented as files. This condition - # checks whether we should yield an empty list (if it is an empty - # directory) or just yield the file itself. - if len(files) == 1 and files[0] == with_prefix and self.isdir(path): - return None - - yield from files diff --git a/dvc/fs/gdrive.py b/dvc/fs/gdrive.py index ca7ae4b6d..8fdae8f0a 100644 --- a/dvc/fs/gdrive.py +++ b/dvc/fs/gdrive.py @@ -8,7 +8,7 @@ from dvc.exceptions import DvcException from dvc.scheme import Schemes from dvc.utils import format_link, tmp_fname -from .fsspec_wrapper import FSSpecWrapper +from .base import FileSystem logger = logging.getLogger(__name__) FOLDER_MIME_TYPE = "application/vnd.google-apps.folder" @@ -30,13 +30,12 @@ class GDriveAuthError(DvcException): super().__init__(message) -class GDriveFileSystem(FSSpecWrapper): # pylint:disable=abstract-method +class GDriveFileSystem(FileSystem): # pylint:disable=abstract-method scheme = Schemes.GDRIVE PARAM_CHECKSUM = "checksum" REQUIRES = {"pydrive2": "pydrive2"} # Always prefer traverse for GDrive since API usage quotas are a concern. TRAVERSE_WEIGHT_MULTIPLIER = 1 - TRAVERSE_PREFIX_LEN = 2 GDRIVE_CREDENTIALS_DATA = "GDRIVE_CREDENTIALS_DATA" DEFAULT_USER_CREDENTIALS_FILE = "gdrive-user-credentials.json" diff --git a/dvc/fs/git.py b/dvc/fs/git.py index 314513a0b..c3ff051f1 100644 --- a/dvc/fs/git.py +++ b/dvc/fs/git.py @@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, Any from funcy import cached_property, wrap_prop -from .fsspec_wrapper import FSSpecWrapper +from .base import FileSystem if TYPE_CHECKING: from scmrepo.fs import GitFileSystem as FsspecGitFileSystem @@ -12,7 +12,7 @@ if TYPE_CHECKING: from scmrepo.git.objects import GitTrie -class GitFileSystem(FSSpecWrapper): # pylint:disable=abstract-method +class GitFileSystem(FileSystem): # pylint:disable=abstract-method """Proxies the repo file access methods to Git objects""" sep = os.sep diff --git a/dvc/fs/gs.py b/dvc/fs/gs.py index 56f17340f..e3eef53f1 100644 --- a/dvc/fs/gs.py +++ b/dvc/fs/gs.py @@ -5,10 +5,10 @@ from funcy import cached_property, wrap_prop from dvc.scheme import Schemes # pylint:disable=abstract-method -from .fsspec_wrapper import ObjectFSWrapper +from .base import ObjectFileSystem -class GSFileSystem(ObjectFSWrapper): +class GSFileSystem(ObjectFileSystem): scheme = Schemes.GS REQUIRES = {"gcsfs": "gcsfs"} PARAM_CHECKSUM = "etag" diff --git a/dvc/fs/hdfs.py b/dvc/fs/hdfs.py index 1b7c4c3dc..163b32691 100644 --- a/dvc/fs/hdfs.py +++ b/dvc/fs/hdfs.py @@ -8,13 +8,13 @@ from funcy import cached_property, wrap_prop from dvc.scheme import Schemes from dvc.utils import fix_env -from .fsspec_wrapper import FSSpecWrapper +from .base import FileSystem CHECKSUM_REGEX = re.compile(r".*\t.*\t(?P.*)") # pylint: disable=abstract-method -class HDFSFileSystem(FSSpecWrapper): +class HDFSFileSystem(FileSystem): scheme = Schemes.HDFS REQUIRES = {"fsspec": "fsspec", "pyarrow": "pyarrow"} PARAM_CHECKSUM = "checksum" diff --git a/dvc/fs/http.py b/dvc/fs/http.py index f44a7c5d5..dbca69263 100644 --- a/dvc/fs/http.py +++ b/dvc/fs/http.py @@ -7,7 +7,7 @@ from dvc import prompt from dvc.scheme import Schemes from ._callback import DEFAULT_CALLBACK -from .fsspec_wrapper import AnyFSPath, FSSpecWrapper +from .base import AnyFSPath, FileSystem @wrap_with(threading.Lock()) @@ -33,7 +33,7 @@ def make_context(ssl_verify): # pylint: disable=abstract-method -class HTTPFileSystem(FSSpecWrapper): +class HTTPFileSystem(FileSystem): scheme = Schemes.HTTP PARAM_CHECKSUM = "checksum" REQUIRES = {"aiohttp": "aiohttp", "aiohttp-retry": "aiohttp_retry"} diff --git a/dvc/fs/local.py b/dvc/fs/local.py index 658f2f90c..a28e2cd8b 100644 --- a/dvc/fs/local.py +++ b/dvc/fs/local.py @@ -10,7 +10,7 @@ from dvc.system import System from dvc.utils import tmp_fname from dvc.utils.fs import copy_fobj_to_file, copyfile, makedirs, move, remove -from .fsspec_wrapper import FSSpecWrapper +from .base import FileSystem logger = logging.getLogger(__name__) @@ -156,7 +156,7 @@ class FsspecLocalFileSystem(AbstractFileSystem): return System.reflink(path1, path2) -class LocalFileSystem(FSSpecWrapper): +class LocalFileSystem(FileSystem): sep = os.sep scheme = Schemes.LOCAL diff --git a/dvc/fs/memory.py b/dvc/fs/memory.py index 32d4402a0..436cdac36 100644 --- a/dvc/fs/memory.py +++ b/dvc/fs/memory.py @@ -4,10 +4,10 @@ from funcy import cached_property, wrap_prop from dvc.scheme import Schemes -from .fsspec_wrapper import FSSpecWrapper +from .base import FileSystem -class MemoryFileSystem(FSSpecWrapper): # pylint:disable=abstract-method +class MemoryFileSystem(FileSystem): # pylint:disable=abstract-method scheme = Schemes.MEMORY PARAM_CHECKSUM = "md5" TRAVERSE_PREFIX_LEN = 2 @@ -18,7 +18,7 @@ class MemoryFileSystem(FSSpecWrapper): # pylint:disable=abstract-method # single global store return isinstance(other, type(self)) - __hash__ = FSSpecWrapper.__hash__ + __hash__ = FileSystem.__hash__ @wrap_prop(threading.Lock()) @cached_property diff --git a/dvc/fs/oss.py b/dvc/fs/oss.py index 22e23d4c5..7bcfdf0d1 100644 --- a/dvc/fs/oss.py +++ b/dvc/fs/oss.py @@ -6,13 +6,13 @@ from funcy import cached_property, wrap_prop from dvc.scheme import Schemes -from .fsspec_wrapper import ObjectFSWrapper +from .base import ObjectFileSystem logger = logging.getLogger(__name__) # pylint:disable=abstract-method -class OSSFileSystem(ObjectFSWrapper): +class OSSFileSystem(ObjectFileSystem): scheme = Schemes.OSS REQUIRES = {"ossfs": "ossfs"} PARAM_CHECKSUM = "etag" diff --git a/dvc/fs/repo.py b/dvc/fs/repo.py index 1de35f3aa..a7d48313e 100644 --- a/dvc/fs/repo.py +++ b/dvc/fs/repo.py @@ -11,7 +11,6 @@ from funcy import cached_property, wrap_prop, wrap_with from ._callback import DEFAULT_CALLBACK from .base import FileSystem from .dvc import DvcFileSystem -from .fsspec_wrapper import FSSpecWrapper from .path import Path if TYPE_CHECKING: @@ -421,7 +420,7 @@ class _RepoFileSystem(AbstractFileSystem): # pylint:disable=abstract-method return dvc_fs.checksum(dvc_path) -class RepoFileSystem(FSSpecWrapper): +class RepoFileSystem(FileSystem): scheme = "local" PARAM_CHECKSUM = "md5" diff --git a/dvc/fs/s3.py b/dvc/fs/s3.py index 7eb9632de..a2cb5fd02 100644 --- a/dvc/fs/s3.py +++ b/dvc/fs/s3.py @@ -8,13 +8,13 @@ from funcy import cached_property, wrap_prop from dvc.scheme import Schemes from ._callback import DEFAULT_CALLBACK -from .fsspec_wrapper import ObjectFSWrapper +from .base import ObjectFileSystem _AWS_CONFIG_PATH = os.path.join(os.path.expanduser("~"), ".aws", "config") # pylint:disable=abstract-method -class BaseS3FileSystem(ObjectFSWrapper): +class BaseS3FileSystem(ObjectFileSystem): scheme = Schemes.S3 REQUIRES = {"s3fs": "s3fs", "boto3": "boto3"} PARAM_CHECKSUM = "etag" diff --git a/dvc/fs/ssh.py b/dvc/fs/ssh.py index ba069b1d1..f9a23428e 100644 --- a/dvc/fs/ssh.py +++ b/dvc/fs/ssh.py @@ -9,7 +9,7 @@ from dvc.scheme import Schemes from dvc.utils.fs import as_atomic from ._callback import DEFAULT_CALLBACK -from .fsspec_wrapper import FSSpecWrapper +from .base import FileSystem _SSH_TIMEOUT = 60 * 30 _SSH_CONFIG_FILE = os.path.expanduser(os.path.join("~", ".ssh", "config")) @@ -27,7 +27,7 @@ def ask_password(host, user, port): # pylint:disable=abstract-method -class SSHFileSystem(FSSpecWrapper): +class SSHFileSystem(FileSystem): scheme = Schemes.SSH REQUIRES = {"sshfs": "sshfs"} diff --git a/dvc/fs/webdav.py b/dvc/fs/webdav.py index 4fb8a766b..fb643014d 100644 --- a/dvc/fs/webdav.py +++ b/dvc/fs/webdav.py @@ -5,17 +5,15 @@ from funcy import cached_property, wrap_prop from dvc.scheme import Schemes -from .fsspec_wrapper import FSSpecWrapper +from .base import FileSystem from .http import ask_password logger = logging.getLogger(__name__) -class WebDAVFileSystem(FSSpecWrapper): # pylint:disable=abstract-method +class WebDAVFileSystem(FileSystem): # pylint:disable=abstract-method scheme = Schemes.WEBDAV root_marker = "" - CAN_TRAVERSE = True - TRAVERSE_PREFIX_LEN = 2 REQUIRES = {"webdav4": "webdav4"} PARAM_CHECKSUM = "etag" diff --git a/dvc/fs/webhdfs.py b/dvc/fs/webhdfs.py index a862867b3..f7661a690 100644 --- a/dvc/fs/webhdfs.py +++ b/dvc/fs/webhdfs.py @@ -5,10 +5,10 @@ from funcy import cached_property, wrap_prop from dvc.scheme import Schemes # pylint:disable=abstract-method -from .fsspec_wrapper import FSSpecWrapper +from .base import FileSystem -class WebHDFSFileSystem(FSSpecWrapper): +class WebHDFSFileSystem(FileSystem): scheme = Schemes.WEBHDFS REQUIRES = {"fsspec": "fsspec"} PARAM_CHECKSUM = "checksum" diff --git a/dvc/ignore.py b/dvc/ignore.py index a1640d9f0..9632804c8 100644 --- a/dvc/ignore.py +++ b/dvc/ignore.py @@ -7,11 +7,11 @@ from itertools import chain, groupby, takewhile from pathspec.patterns import GitWildMatchPattern from pathspec.util import normalize_file -from dvc.fs.base import FileSystem +from dvc.fs.base import AnyFSPath, FileSystem from dvc.fs.local import localfs from dvc.pathspec_math import PatternInfo, merge_patterns from dvc.scheme import Schemes -from dvc.types import AnyPath, List, Optional +from dvc.types import List, Optional from dvc.utils.collections import PathStringTrie logger = logging.getLogger(__name__) @@ -287,7 +287,7 @@ class DvcIgnoreFilter: return [fs_dict[name] for name in chain(dirs, nondirs)] - def walk(self, fs: FileSystem, path: AnyPath, **kwargs): + def walk(self, fs: FileSystem, path: AnyFSPath, **kwargs): ignore_subrepos = kwargs.pop("ignore_subrepos", True) if fs.scheme == Schemes.LOCAL: for root, dirs, files in fs.walk(path, **kwargs): @@ -298,7 +298,7 @@ class DvcIgnoreFilter: else: yield from fs.walk(path, **kwargs) - def find(self, fs: FileSystem, path: AnyPath, **kwargs): + def find(self, fs: FileSystem, path: AnyFSPath, **kwargs): if fs.scheme == Schemes.LOCAL: for root, _, files in self.walk(fs, path, **kwargs): for file in files: diff --git a/dvc/objects/db.py b/dvc/objects/db.py index 4336b9bb4..feeadd585 100644 --- a/dvc/objects/db.py +++ b/dvc/objects/db.py @@ -12,9 +12,8 @@ from dvc.progress import Tqdm if TYPE_CHECKING: from typing import Tuple - from dvc.fs.base import FileSystem + from dvc.fs.base import AnyFSPath, FileSystem from dvc.hash_info import HashInfo - from dvc.types import AnyPath logger = logging.getLogger(__name__) @@ -78,8 +77,8 @@ class ObjectDB: def _add_file( self, from_fs: "FileSystem", - from_info: "AnyPath", - to_info: "AnyPath", + from_info: "AnyFSPath", + to_info: "AnyFSPath", _hash_info: "HashInfo", hardlink: bool = False, ): @@ -96,7 +95,7 @@ class ObjectDB: def add( self, - fs_path: "AnyPath", + fs_path: "AnyFSPath", fs: "FileSystem", hash_info: "HashInfo", hardlink: bool = False, diff --git a/dvc/objects/file.py b/dvc/objects/file.py index eeeb45aaa..6ede595fb 100644 --- a/dvc/objects/file.py +++ b/dvc/objects/file.py @@ -4,9 +4,8 @@ import os from typing import TYPE_CHECKING, Optional if TYPE_CHECKING: - from dvc.fs.base import FileSystem + from dvc.fs.base import AnyFSPath, FileSystem from dvc.hash_info import HashInfo - from dvc.types import AnyPath from .db import ObjectDB @@ -16,7 +15,7 @@ logger = logging.getLogger(__name__) class HashFile: def __init__( self, - fs_path: Optional["AnyPath"], + fs_path: "AnyFSPath", fs: Optional["FileSystem"], hash_info: "HashInfo", name: Optional[str] = None, diff --git a/dvc/repo/collect.py b/dvc/repo/collect.py index 58ba06e61..a73b1b6b2 100644 --- a/dvc/repo/collect.py +++ b/dvc/repo/collect.py @@ -30,14 +30,14 @@ def _collect_paths( targets: Iterable[str], recursive: bool = False, rev: str = None, -): +) -> StrPaths: from dvc.fs.repo import RepoFileSystem from dvc.utils import relpath fs_paths = [os.path.abspath(target) for target in targets] fs = RepoFileSystem(repo=repo) - target_paths = [] + target_paths: StrPaths = [] for fs_path in fs_paths: if recursive and fs.isdir(fs_path): target_paths.extend(fs.find(fs_path)) diff --git a/dvc/repo/experiments/executor/ssh.py b/dvc/repo/experiments/executor/ssh.py index 657e05b96..e57d3a1ec 100644 --- a/dvc/repo/experiments/executor/ssh.py +++ b/dvc/repo/experiments/executor/ssh.py @@ -174,6 +174,7 @@ class SSHExecutor(BaseExecutor): script_path = self._setup_script_path( posixpath.join(self._repo_abspath, self.dvc_dir) ) + assert self._setup_script fs.upload(self._setup_script, script_path) def _ssh_cmd(self, sshfs, cmd, chdir=None, **kwargs): diff --git a/tests/unit/remote/test_base.py b/tests/unit/remote/test_base.py index 968e7db70..7c1aa62c0 100644 --- a/tests/unit/remote/test_base.py +++ b/tests/unit/remote/test_base.py @@ -34,16 +34,18 @@ def test_hashes_exist(object_exists, traverse, dvc): # large remote, small local object_exists.reset_mock() traverse.reset_mock() - with mock.patch.object(odb, "_list_hashes", return_value=list(range(256))): + with mock.patch.object( + odb, "_list_hashes", return_value=list(range(2048)) + ): hashes = list(range(1000)) odb.hashes_exist(hashes) # verify that _odb_paths_with_max() short circuits - # before returning all 256 remote hashes + # before returning all 2048 remote hashes max_hashes = math.ceil( odb._max_estimation_size(hashes) / pow(16, odb.fs.TRAVERSE_PREFIX_LEN) ) - assert max_hashes < 256 + assert max_hashes < 2048 object_exists.assert_called_with( frozenset(range(max_hashes, 1000)), None, None ) @@ -75,7 +77,7 @@ def test_list_hashes_traverse(_path_to_hash, list_hashes, dvc): size = 256 / odb.fs._JOBS * odb.fs.LIST_OBJECT_PAGE_SIZE list(odb._list_hashes_traverse(size, {0})) for i in range(1, 16): - list_hashes.assert_any_call(f"{i:03x}") + list_hashes.assert_any_call(f"{i:0{odb.fs.TRAVERSE_PREFIX_LEN}x}") for i in range(1, 256): list_hashes.assert_any_call(f"{i:02x}")