fs: make localfs fsspec-compliant

It is not completely fsspec compliant, but it's upto the level
required by our FSSpecWrapper.

Next thing to tackle would be to merging `fs.base.FileSystem`
and `fs.fsspec_wrapper.FSSpecWrapper` into one filesystem
that is lazily-loaded and is itself fsspec-compliant.

dvc/fs/fsspec_wrapper.py

@@ -4,7 +4,7 @@ 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
+from .base import FileSystem, RemoteActionNotImplemented
 
 FSPath = str
 AnyFSPath = str
@@ -66,7 +66,7 @@ class FSSpecWrapper(FileSystem):
         encoding: Optional[str] = None,
         **kwargs,
     ) -> "IO":  # pylint: disable=arguments-differ
-        return self.fs.open(path, mode=mode)
+        return self.fs.open(path, mode=mode, encoding=encoding, **kwargs)
 
     def checksum(self, path: AnyFSPath) -> str:
         return self.fs.checksum(path)
@@ -78,6 +78,42 @@ class FSSpecWrapper(FileSystem):
     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]"
@@ -88,7 +124,7 @@ class FSSpecWrapper(FileSystem):
     def ls(self, path: AnyFSPath, detail: "Literal[False]") -> Iterator[str]:
         ...
 
-    def ls(self, path, detail=False):
+    def ls(self, path, detail=False, **kwargs):
         return self.fs.ls(path, detail=detail)
 
     def find(self, path, prefix=None):

dvc/fs/local.py

@@ -1,43 +1,45 @@
 import logging
 import os
+import threading
 
-from funcy import cached_property
+from fsspec import AbstractFileSystem
+from funcy import cached_property, wrap_prop
 
 from dvc.scheme import Schemes
 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 ._callback import DEFAULT_CALLBACK
-from .base import FileSystem
+from .fsspec_wrapper import FSSpecWrapper
 
 logger = logging.getLogger(__name__)
 
 
-class LocalFileSystem(FileSystem):
+# pylint:disable=abstract-method, arguments-differ
+class FsspecLocalFileSystem(AbstractFileSystem):
     sep = os.sep
 
-    scheme = Schemes.LOCAL
-    PARAM_CHECKSUM = "md5"
-    PARAM_PATH = "path"
-    TRAVERSE_PREFIX_LEN = 2
-
-    def __init__(self, **config):
+    def __init__(self, *args, **kwargs):
         from fsspec.implementations.local import LocalFileSystem as LocalFS
 
-        super().__init__(**config)
+        super().__init__(*args, **kwargs)
         self.fs = LocalFS()
 
-    @cached_property
-    def path(self):
-        from .path import Path
+    def makedirs(self, path, exist_ok=False):
+        makedirs(path, exist_ok=exist_ok)
 
-        return Path(self.sep, os.getcwd)
+    def mkdir(self, path, create_parents=True, **kwargs):
+        if self.exists(path):
+            raise FileExistsError(path)
+        if create_parents:
+            self.makedirs(path, exist_ok=True)
+        else:
+            os.mkdir(path, **kwargs)
 
-    def open(self, path, mode="r", encoding=None, **kwargs):
-        return open(path, mode=mode, encoding=encoding)
+    def lexists(self, path, **kwargs):
+        return os.path.lexists(path)
 
-    def exists(self, path) -> bool:
+    def exists(self, path, **kwargs):
         # TODO: replace this with os.path.exists once the problem is fixed on
         # the fsspec https://github.com/intake/filesystem_spec/issues/742
         return os.path.lexists(path)
@@ -46,82 +48,89 @@ class LocalFileSystem(FileSystem):
         from fsspec.utils import tokenize
 
         st = os.stat(path)
-
         return str(int(tokenize([st.st_ino, st.st_mtime, st.st_size]), 16))
 
+    def info(self, path, **kwargs):
+        return self.fs.info(path)
+
+    def ls(self, path, **kwargs):
+        return self.fs.ls(path, **kwargs)
+
     def isfile(self, path) -> bool:
         return os.path.isfile(path)
 
     def isdir(self, path) -> bool:
         return os.path.isdir(path)
 
-    def iscopy(self, path):
-        return not (System.is_symlink(path) or System.is_hardlink(path))
-
-    def walk(self, top, topdown=True, **kwargs):
+    def walk(self, path, maxdepth=None, topdown=True, **kwargs):
         """Directory fs generator.
 
         See `os.walk` for the docs. Differences:
         - no support for symlinks
         """
         for root, dirs, files in os.walk(
-            top,
+            path,
             topdown=topdown,
         ):
             yield os.path.normpath(root), dirs, files
 
-    def find(self, path, prefix=None):
-        for root, _, files in self.walk(path):
+    def find(self, path, **kwargs):
+        for root, _, files in self.walk(path, **kwargs):
             for file in files:
                 # NOTE: os.path.join is ~5.5 times slower
                 yield f"{root}{os.sep}{file}"
 
-    def is_empty(self, path):
-        if self.isfile(path) and os.path.getsize(path) == 0:
-            return True
+    @classmethod
+    def _parent(cls, path):
+        return os.path.dirname(path)
 
-        if self.isdir(path) and len(os.listdir(path)) == 0:
-            return True
+    def put_file(self, lpath, rpath, callback=None, **kwargs):
+        parent = self._parent(rpath)
+        makedirs(parent, exist_ok=True)
+        tmp_file = os.path.join(parent, tmp_fname())
+        copyfile(lpath, tmp_file, callback=callback)
+        os.replace(tmp_file, rpath)
 
-        return False
+    def get_file(self, rpath, lpath, callback=None, **kwargs):
+        copyfile(rpath, lpath, callback=callback)
 
-    def remove(self, path):
+    def mv(self, path1, path2, **kwargs):
+        self.makedirs(self._parent(path2), exist_ok=True)
+        move(path1, path2)
+
+    def rmdir(self, path):
+        os.rmdir(path)
+
+    def rm_file(self, path):
         remove(path)
 
-    def makedirs(self, path, **kwargs):
-        makedirs(path, exist_ok=kwargs.pop("exist_ok", True))
+    def rm(self, path, recursive=False, maxdepth=None):
+        remove(path)
 
-    def move(self, from_info, to_info):
-        self.makedirs(self.path.parent(to_info))
-        move(from_info, to_info)
-
-    def copy(self, from_info, to_info):
-        tmp_info = self.path.join(self.path.parent(to_info), tmp_fname(""))
+    def copy(self, path1, path2, recursive=False, on_error=None, **kwargs):
+        tmp_info = os.path.join(self._parent(path2), tmp_fname(""))
         try:
-            copyfile(from_info, tmp_info)
-            os.rename(tmp_info, to_info)
+            copyfile(path1, tmp_info)
+            os.rename(tmp_info, path2)
         except Exception:
-            self.remove(tmp_info)
+            self.rm_file(tmp_info)
             raise
 
-    def upload_fobj(self, fobj, to_info, **kwargs):
-        self.makedirs(self.path.parent(to_info))
-        tmp_info = self.path.join(self.path.parent(to_info), tmp_fname(""))
-        try:
-            copy_fobj_to_file(fobj, tmp_info)
-            os.rename(tmp_info, to_info)
-        except Exception:
-            self.remove(tmp_info)
-            raise
+    def open(self, path, mode="r", encoding=None, **kwargs):
+        return open(path, mode=mode, encoding=encoding)
 
-    def symlink(self, from_info, to_info):
-        System.symlink(from_info, to_info)
+    def symlink(self, path1, path2):
+        return System.symlink(path1, path2)
 
     @staticmethod
     def is_symlink(path):
         return System.is_symlink(path)
 
-    def hardlink(self, from_info, to_info):
+    @staticmethod
+    def is_hardlink(path):
+        return System.is_hardlink(path)
+
+    def hardlink(self, path1, path2):
         # If there are a lot of empty files (which happens a lot in datasets),
         # and the cache type is `hardlink`, we might reach link limits and
         # will get something like: `too many links error`
@@ -135,40 +144,46 @@ class LocalFileSystem(FileSystem):
         #   * Windows with NTFS has a limit of 1024 hard links on a file
         #
         # That's why we simply create an empty file rather than a link.
-        if self.getsize(from_info) == 0:
-            self.open(to_info, "w").close()
+        if self.size(path1) == 0:
+            self.open(path2, "w").close()
 
-            logger.debug("Created empty file: %s -> %s", from_info, to_info)
+            logger.debug("Created empty file: %s -> %s", path1, path2)
             return
 
-        System.hardlink(from_info, to_info)
+        return System.hardlink(path1, path2)
 
-    @staticmethod
-    def is_hardlink(path):
-        return System.is_hardlink(path)
+    def reflink(self, path1, path2):
+        return System.reflink(path1, path2)
 
-    def reflink(self, from_info, to_info):
-        System.reflink(from_info, to_info)
 
-    def info(self, path):
-        return self.fs.info(path)
+class LocalFileSystem(FSSpecWrapper):
+    sep = os.sep
 
-    def ls(self, path, **kwargs):
-        return self.fs.ls(path, **kwargs)
+    scheme = Schemes.LOCAL
+    PARAM_CHECKSUM = "md5"
+    PARAM_PATH = "path"
+    TRAVERSE_PREFIX_LEN = 2
 
-    def put_file(
-        self, from_file, to_info, callback=DEFAULT_CALLBACK, **kwargs
-    ):
-        parent = self.path.parent(to_info)
-        makedirs(parent, exist_ok=True)
-        tmp_file = self.path.join(parent, tmp_fname())
-        copyfile(from_file, tmp_file, callback=callback)
-        os.replace(tmp_file, to_info)
+    @wrap_prop(threading.Lock())
+    @cached_property
+    def fs(self):
+        return FsspecLocalFileSystem(**self.config)
 
-    def get_file(
-        self, from_info, to_file, callback=DEFAULT_CALLBACK, **kwargs
-    ):
-        copyfile(from_info, to_file, callback=callback)
+    @cached_property
+    def path(self):
+        from .path import Path
+
+        return Path(self.sep, os.getcwd)
+
+    def upload_fobj(self, fobj, to_info, **kwargs):
+        self.makedirs(self.path.parent(to_info))
+        tmp_info = self.path.join(self.path.parent(to_info), tmp_fname(""))
+        try:
+            copy_fobj_to_file(fobj, tmp_info)
+            os.rename(tmp_info, to_info)
+        except Exception:
+            self.remove(tmp_info)
+            raise
 
 
 localfs = LocalFileSystem()

tests/func/test_add.py

@@ -783,7 +783,7 @@ def test_add_optimization_for_hardlink_on_empty_files(tmp_dir, dvc, mocker):
     m = mocker.spy(LocalFileSystem, "is_hardlink")
     stages = dvc.add(["foo", "bar", "lorem", "ipsum"])
 
-    assert m.call_count == 4
+    assert m.call_count == 8
     assert m.call_args != call(tmp_dir / "foo")
     assert m.call_args != call(tmp_dir / "bar")