Repository: fsspec/filesystem_spec
Pull request #1092 (2022-11-02 15:29:42)

cache: added clear_expired_cache (#1055 - #1061)
As discussed in #1061 based on the discussion #1055 time to time there is the need to keep data locally and remove them after a defined time.
.clear_expired_cache() add the possibility to remove only the cached data that overpass the defined time.



Patch:
diff --git a/docs/environment.yml b/docs/environment.yml
index b5f5a9d..814bd85 100644
--- a/docs/environment.yml
+++ b/docs/environment.yml
@@ -5,3 +5,4 @@ dependencies:
   - python=3.9
   - docutils<0.17
   - numpydoc
+  - yarl
diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index 8d8c428..0edf8c5 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -144,9 +144,12 @@ class ArrowFSWrapper(AbstractFileSystem):
             self.fs.delete_file(path)
 
     @wrap_exceptions
-    def _open(self, path, mode="rb", block_size=None, **kwargs):
+    def _open(self, path, mode="rb", block_size=None, seekable=False, **kwargs):
         if mode == "rb":
-            method = self.fs.open_input_stream
+            if seekable:
+                method = self.fs.open_input_file
+            else:
+                method = self.fs.open_input_stream
         elif mode == "wb":
             method = self.fs.open_output_stream
         elif mode == "ab":
@@ -155,9 +158,10 @@ class ArrowFSWrapper(AbstractFileSystem):
             raise ValueError(f"unsupported mode for Arrow filesystem: {mode!r}")
 
         _kwargs = {}
-        if PYARROW_VERSION[0] >= 4:
-            # disable compression auto-detection
-            _kwargs["compression"] = None
+        if mode != "rb" or not seekable:
+            if PYARROW_VERSION[0] >= 4:
+                # disable compression auto-detection
+                _kwargs["compression"] = None
         stream = method(path, **_kwargs)
 
         return ArrowFile(self, stream, path, mode, block_size, **kwargs)
@@ -182,7 +186,7 @@ class ArrowFSWrapper(AbstractFileSystem):
 
 
 @mirror_from(
-    "stream", ["read", "seek", "tell", "write", "readable", "writable", "close"]
+    "stream", ["read", "seek", "tell", "write", "readable", "writable", "close", "size"]
 )
 class ArrowFile(io.IOBase):
     def __init__(self, fs, stream, path, mode, block_size=None, **kwargs):
diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index a0799a0..0619d1a 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -232,6 +232,44 @@ class CachingFileSystem(AbstractFileSystem):
         rmtree(self.storage[-1])
         self.load_cache()
 
+    def clear_expired_cache(self, expiry_time=None):
+        """Remove all expired files and metadata from the cache
+
+        In the case of multiple cache locations, this clears only the last one,
+        which is assumed to be the read/write one.
+
+        Parameters
+        ----------
+        expiry_time: int
+            The time in seconds after which a local copy is considered useless.
+            If not defined the default is equivalent to the attribute from the
+            file caching instantiation.
+        """
+
+        if not expiry_time:
+            expiry_time = self.expiry
+
+        self._check_cache()
+
+        for path, detail in self.cached_files[-1].copy().items():
+            if time.time() - detail["time"] > expiry_time:
+                if self.same_names:
+                    basename = os.path.basename(detail["original"])
+                    fn = os.path.join(self.storage[-1], basename)
+                else:
+                    fn = os.path.join(self.storage[-1], detail["fn"])
+                if os.path.exists(fn):
+                    os.remove(fn)
+                    self.cached_files[-1].pop(path)
+
+        if self.cached_files[-1]:
+            cache_path = os.path.join(self.storage[-1], "cache")
+            with open(cache_path, "wb") as fc:
+                pickle.dump(self.cached_files[-1], fc)
+        else:
+            rmtree(self.storage[-1])
+            self.load_cache()
+
     def pop_from_cache(self, path):
         """Remove cached version of given file
 
@@ -389,6 +427,7 @@ class CachingFileSystem(AbstractFileSystem):
             "_check_cache",
             "_mkcache",
             "clear_cache",
+            "clear_expired_cache",
             "pop_from_cache",
             "_mkcache",
             "local_file",


Test patch:
diff --git a/fsspec/implementations/tests/test_arrow.py b/fsspec/implementations/tests/test_arrow.py
index 8aca6ea..1386df8 100644
--- a/fsspec/implementations/tests/test_arrow.py
+++ b/fsspec/implementations/tests/test_arrow.py
@@ -194,3 +194,14 @@ def test_open_append(fs, remote_dir):
 
     with fs.open(remote_dir + "/a.txt") as stream:
         assert stream.read() == 2 * data
+
+
+def test_open_seekable(fs, remote_dir):
+    data = b"dvc.org"
+
+    with fs.open(remote_dir + "/a.txt", "wb") as stream:
+        stream.write(data)
+
+    with fs.open(remote_dir + "/a.txt", "rb", seekable=True) as file:
+        file.seek(2)
+        assert file.read() == data[2:]
diff --git a/fsspec/implementations/tests/test_cached.py b/fsspec/implementations/tests/test_cached.py
index 4ddc051..51c451e 100644
--- a/fsspec/implementations/tests/test_cached.py
+++ b/fsspec/implementations/tests/test_cached.py
@@ -153,6 +153,109 @@ def test_clear():
     assert len(os.listdir(cache1)) < 2
 
 
+def test_clear_expired(tmp_path):
+    def __ager(cache_fn, fn):
+        """
+        Modify the cache file to virtually add time lag to selected files.
+
+        Parameters
+        ---------
+        cache_fn: str
+            cache path
+        fn: str
+            file name to be modified
+        """
+        import pathlib
+        import time
+
+        if os.path.exists(cache_fn):
+            with open(cache_fn, "rb") as f:
+                cached_files = pickle.load(f)
+                fn_posix = pathlib.Path(fn).as_posix()
+                cached_files[fn_posix]["time"] = cached_files[fn_posix]["time"] - 691200
+            assert os.access(cache_fn, os.W_OK), "Cache is not writable"
+            with open(cache_fn, "wb") as f:
+                pickle.dump(cached_files, f)
+            time.sleep(1)
+
+    origin = tmp_path.joinpath("origin")
+    cache1 = tmp_path.joinpath("cache1")
+    cache2 = tmp_path.joinpath("cache2")
+    cache3 = tmp_path.joinpath("cache3")
+
+    origin.mkdir()
+    cache1.mkdir()
+    cache2.mkdir()
+    cache3.mkdir()
+
+    data = b"test data"
+    f1 = origin.joinpath("afile")
+    f2 = origin.joinpath("bfile")
+    f3 = origin.joinpath("cfile")
+    f4 = origin.joinpath("dfile")
+
+    with open(f1, "wb") as f:
+        f.write(data)
+    with open(f2, "wb") as f:
+        f.write(data)
+    with open(f3, "wb") as f:
+        f.write(data)
+    with open(f4, "wb") as f:
+        f.write(data)
+
+    # populates first cache
+    fs = fsspec.filesystem(
+        "filecache", target_protocol="file", cache_storage=str(cache1), cache_check=1
+    )
+    assert fs.cat(str(f1)) == data
+
+    # populates "last" cache if file not found in first one
+    fs = fsspec.filesystem(
+        "filecache",
+        target_protocol="file",
+        cache_storage=[str(cache1), str(cache2)],
+        cache_check=1,
+    )
+    assert fs.cat(str(f2)) == data
+    assert fs.cat(str(f3)) == data
+    assert len(os.listdir(cache2)) == 3
+
+    # force the expiration
+    cache_fn = os.path.join(fs.storage[-1], "cache")
+    __ager(cache_fn, f2)
+
+    # remove from cache2 the expired files
+    fs.clear_expired_cache()
+    assert len(os.listdir(cache2)) == 2
+
+    # check complete cleanup
+    __ager(cache_fn, f3)
+
+    fs.clear_expired_cache()
+    assert not fs._check_file(f2)
+    assert not fs._check_file(f3)
+    assert len(os.listdir(cache2)) < 2
+
+    # check cache1 to be untouched after cleaning
+    assert len(os.listdir(cache1)) == 2
+
+    # check cleaning with 'same_name' option enabled
+    fs = fsspec.filesystem(
+        "filecache",
+        target_protocol="file",
+        cache_storage=[str(cache1), str(cache2), str(cache3)],
+        same_names=True,
+        cache_check=1,
+    )
+    assert fs.cat(str(f4)) == data
+
+    cache_fn = os.path.join(fs.storage[-1], "cache")
+    __ager(cache_fn, f4)
+
+    fs.clear_expired_cache()
+    assert not fs._check_file(str(f4))
+
+
 def test_pop():
     import tempfile