Merge branch 'mr/ramonat/support-iterators-in-fs' into 'master'

Support iterators in mv, ls, and rm

Closes #26

See merge request it/e3-core!62

src/e3/fs.py

@@ -259,7 +259,7 @@ def get_content(file_path: str) -> bytes:
 
 
 def ls(
-    path: str | list[str] | Path | list[Path], emit_log_record: bool = True
+    path: str | Iterable[str] | Path | Iterable[Path], emit_log_record: bool = True
 ) -> list[str]:
     """list files.
 
@@ -312,11 +312,11 @@ def mkdir(path: str | Path, mode: int = 0o755, quiet: bool = False) -> None:
             raise FSError(origin="mkdir", message=f"can't create {path}") from e
 
 
-def mv(source: str | Path | list[str] | list[Path], target: str | Path) -> None:
+def mv(source: str | Path | Iterable[str] | Iterable[Path], target: str | Path) -> None:
     """Move files.
 
-    :param source: a glob pattern, a list of glob patterns, a path, or a list
-        of paths
+    :param source: a glob pattern, a sequence/iterator of glob patterns, a path,
+        or a sequence/iterator of paths
     :param target: target file or directory. If the source resolves as
         several files then target should be a directory
 
@@ -331,6 +331,7 @@ def move_file(src: str, dst: str) -> None:
         rmtree. This ensure moving a directory with read-only files will
         work.
         """
+        logger.debug(f"mv {src} {dst}")
 
         def same_file(src: str, dst: str) -> bool:
             if hasattr(os.path, "samefile"):
@@ -388,13 +389,6 @@ def destinsrc(src: str, dst: str) -> bool:
                 rm(src)
         return
 
-    if isinstance(source, (str, Path)):
-        logger.debug(f"mv {os.fspath(source)} {os.fspath(target)}")
-    else:
-        logger.debug(
-            f"mv {' '.join([os.fspath(s) for s in source])} {os.fspath(target)}"
-        )
-
     try:
         # Compute file list and number of file to copy
         file_list = ls(source, emit_log_record=False)
@@ -414,21 +408,21 @@ def destinsrc(src: str, dst: str) -> bool:
         else:
             for f in file_list:
                 f_dest = os.path.join(target, os.path.basename(f))
-                e3.log.debug("mv %s %s", f, f_dest)
                 move_file(f, f_dest)
     except Exception as e:
         logger.error(e)
         raise FSError(origin="mv", message=str(e)) from e
 
 
 def rm(
-    path: str | Path | list[str] | list[Path],
+    path: str | Path | Iterable[str] | Iterable[Path],
     recursive: bool = False,
     glob: bool = True,
 ) -> None:
     """Remove files.
 
-    :param path: a glob pattern, or a list of glob patterns
+    :param path: a glob pattern, a sequence/iterator of glob patterns, a path,
+        or a sequence/iterator of paths
     :param recursive: if True do a recursive deletion. Default is False
     :param glob: if True globbing pattern expansion is used
 
@@ -437,13 +431,6 @@ def rm(
     Note that the function will not raise an error is there are no file to
     delete.
     """
-    if isinstance(path, (str, Path)):
-        logger.debug(f"rm{' -r' if recursive else ''} {os.fspath(path)}")
-    else:
-        logger.debug(
-            f"rm{' -r' if recursive else ''} {' '.join(os.fspath(p) for p in path)}"
-        )
-
     # We transform the list into a set in order to remove duplicate files in
     # the list
     if glob:
@@ -454,6 +441,8 @@ def rm(
         else:
             file_list = {os.fspath(p) for p in path}
 
+    logger.debug(f"rm{' -r' if recursive else ''} {' '.join(file_list)}")
+
     def onerror(
         func: Callable[..., Any], error_path: str, exc_info: tuple | BaseException
     ) -> None:

tests/tests_e3/fs/main_test.py

@@ -109,6 +109,32 @@ def test_pathlib():
     ]
 
 
+def test_mv_with_iterables():
+    for idx in range(10):
+        e3.os.fs.touch(f"a{idx}")
+
+    def star(d):
+        for idx in range(10):
+            yield f"{d}{idx}"
+
+    e3.fs.mkdir("dst")
+    e3.fs.mv(star("a"), "dst")
+    for idx in range(10):
+        assert os.path.exists(os.path.join("dst", (f"a{idx}")))
+
+    def dst_star(d):
+        for idx in range(10):
+            yield Path("dst") / f"{d}{idx}"
+
+    assert e3.fs.ls(dst_star("a")) == [
+        os.path.join("dst", f"a{idx}") for idx in range(10)
+    ]
+
+    e3.fs.rm(dst_star("a"))
+    for idx in range(10):
+        assert not os.path.exists(os.path.join("dst", (f"a{idx}")))
+
+
 @pytest.mark.skipif(sys.platform == "win32", reason="test using symlink")
 def test_cp_symplink():
     e3.os.fs.touch("c")