Repository: ungarj/mapchete
Pull request #673 (2025-03-25 15:44:47)

enable extended fs-options & allow mpath cp to copy directories


Patch:
diff --git a/mapchete/cli/default/convert.py b/mapchete/cli/default/convert.py
index 40ccc57..08ae3b6 100644
--- a/mapchete/cli/default/convert.py
+++ b/mapchete/cli/default/convert.py
@@ -10,6 +10,7 @@ from mapchete.cli import options
 from mapchete.cli.progress_bar import PBar
 from mapchete.config import DaskSettings
 from mapchete.formats import available_output_formats
+from mapchete.path import MPath
 
 OUTPUT_FORMATS = available_output_formats()
 
@@ -114,6 +115,7 @@ def convert(
     no_pbar=False,
     verbose=False,
     dask_no_task_graph=False,
+    src_fs_opts=None,
     logfile=None,
     **kwargs,
 ):
@@ -122,7 +124,7 @@ def convert(
             total=100, desc="tasks", disable=debug or no_pbar, print_messages=verbose
         ) as pbar:
             commands.convert(
-                tiledir,
+                MPath(tiledir, storage_options=src_fs_opts),
                 output,
                 *args,
                 dask_settings=DaskSettings(process_graph=not dask_no_task_graph),
diff --git a/mapchete/cli/default/index.py b/mapchete/cli/default/index.py
index f95abda..54cde1f 100644
--- a/mapchete/cli/default/index.py
+++ b/mapchete/cli/default/index.py
@@ -1,6 +1,7 @@
 """Create index for process output."""
 
 import logging
+from typing import Dict, Optional
 
 import click
 import tqdm
@@ -8,6 +9,7 @@ import tqdm
 from mapchete import commands
 from mapchete.cli import options
 from mapchete.cli.progress_bar import PBar
+from mapchete.path import MPath
 
 # workaround for https://github.com/tqdm/tqdm/issues/481
 tqdm.monitor_interval = 0
@@ -42,7 +44,16 @@ logger = logging.getLogger(__name__)
 @options.opt_no_pbar
 @options.opt_debug
 @options.opt_logfile
-def index(*args, debug=False, no_pbar=False, verbose=False, logfile=None, **kwargs):
+def index(
+    tiledir: MPath,
+    fs_opts: Optional[Dict] = None,
+    *args,
+    debug=False,
+    no_pbar=False,
+    verbose=False,
+    logfile=None,
+    **kwargs,
+):
     """Create various index files from process output."""
     # handle deprecated options
     for x in ["password", "username"]:
@@ -56,10 +67,10 @@ def index(*args, debug=False, no_pbar=False, verbose=False, logfile=None, **kwar
     with PBar(
         total=100, desc="tiles", disable=debug or no_pbar, print_messages=verbose
     ) as pbar:
-        kwargs.update(some_input=kwargs.pop("tiledir"))
         commands.index(
+            MPath(tiledir, storage_options=fs_opts),
             *args,
             observers=[pbar],
             **kwargs,
         )
-    tqdm.tqdm.write(f"index(es) creation for {kwargs.get('tiledir')} finished")
+    tqdm.tqdm.write(f"index(es) creation for {str(tiledir)} finished")
diff --git a/mapchete/cli/mpath.py b/mapchete/cli/mpath.py
index a32cfa4..abbc86e 100644
--- a/mapchete/cli/mpath.py
+++ b/mapchete/cli/mpath.py
@@ -1,12 +1,17 @@
 import json
+import logging
+import os
 from typing import List, Optional, Union
 
 import click
+import tqdm
 
 from mapchete.cli import options
 from mapchete.io import copy
 from mapchete.path import MPath
 
+logger = logging.getLogger(__name__)
+
 opt_recursive = click.option("--recursive", "-r", is_flag=True)
 
 
@@ -24,30 +29,71 @@ def exists(path: MPath):
 @mpath.command(help="copy path.")
 @options.arg_path
 @options.arg_out_path
+@options.opt_src_fs_opts
+@options.opt_dst_fs_opts
 @options.opt_overwrite
-def cp(path: MPath, out_path: MPath, overwrite: bool = False):
+@opt_recursive
+@click.option("--skip-existing", is_flag=True)
+@options.opt_debug
+def cp(
+    path: MPath,
+    out_path: MPath,
+    overwrite: bool = False,
+    recursive: bool = False,
+    skip_existing: bool = False,
+    debug: bool = False,
+    **_,
+):
     try:
-        copy(path, out_path, overwrite=overwrite)
+        if path.is_directory():
+            if not recursive:  # pragma: no cover
+                raise click.UsageError(
+                    "source path is directory, --recursive flag required"
+                )
+            for _, __, files in tqdm.tqdm(path.walk(), disable=debug):
+                for src_file in tqdm.tqdm(
+                    [file for file in files if not file.is_directory()],
+                    leave=False,
+                    disable=debug,
+                ):
+                    dst_file = out_path / os.path.relpath(
+                        str(src_file.without_protocol()),
+                        start=str(path.without_protocol()),
+                    )
+                    message = f"copy {str(src_file)} to {str(dst_file)} ..."
+                    if debug:  # pragma: no cover
+                        logger.debug(message)
+                    else:
+                        tqdm.tqdm.write(message)
+                    copy(
+                        src_file, dst_file, overwrite=overwrite, exists_ok=skip_existing
+                    )
+        else:
+            copy(path, out_path, overwrite=overwrite, exists_ok=skip_existing)
     except Exception as exc:  # pragma: no cover
-        raise click.ClickException(exc)
+        if debug:
+            raise
+        raise click.ClickException(str(exc))
 
 
 @mpath.command(help="Remove path.")
 @options.arg_path
+@options.opt_src_fs_opts
 @opt_recursive
 @options.opt_force
-def rm(path: MPath, recursive: bool = False, force: bool = False):
+def rm(path: MPath, recursive: bool = False, force: bool = False, **_):
     try:
         if force or click.confirm(
             f"do you really want to permanently delete {str(path)}?"
         ):
             path.rm(recursive=recursive)
     except Exception as exc:  # pragma: no cover
-        raise click.ClickException(exc)
+        raise click.ClickException(str(exc))
 
 
 @mpath.command(help="List path contents.")
 @options.arg_path
+@options.opt_src_fs_opts
 @click.option(
     "--date-format", type=click.STRING, default="%y-%m-%d %H:%M:%S", show_default=True
 )
@@ -62,6 +108,7 @@ def ls(
     spacing: int = 4,
     recursive: bool = False,
     max_depth: Optional[int] = None,
+    **_,
 ):
     size_column_width = 10
 
@@ -114,8 +161,8 @@ def ls(
         if recursive:
             for root, _, files in path.walk(absolute_paths=True, maxdepth=max_depth):
                 _print_rows(
-                    [root],
-                    files,
+                    [root],  # type: ignore
+                    files,  # type: ignore
                     last_modified_column_width=last_modified_column_width,
                     spacing=spacing,
                 )
@@ -123,7 +170,7 @@ def ls(
             directories = []
             files = []
             for subpath in path.ls(absolute_paths=True):
-                if subpath.is_directory():
+                if subpath.is_directory():  # type: ignore
                     directories.append(subpath)
                 else:
                     files.append(subpath)
@@ -135,7 +182,7 @@ def ls(
             )
 
     except Exception as exc:  # pragma: no cover
-        raise click.ClickException(exc)
+        raise click.ClickException(str(exc))
 
 
 def _row(
@@ -149,7 +196,7 @@ def _row(
             width = len(text)
         return text + " " * (width - len(text))
 
-    def _column_underline(text: str = "", width: int = 4, symbol: str = "-"):
+    def _column_underline(text: str = "", width: int = 4, symbol: str = "-") -> str:
         if width is None or len(text) > width:
             width = len(text)
         return symbol * len(text) + " " * (width - len(text))
@@ -161,7 +208,7 @@ def _row(
     if underlines:
         out += "\n"
         out += space.join(
-            [_column_underline(column, width) for column, width in zip(columns, widths)]
+            [_column_underline(column, width) for column, width in zip(columns, widths)]  # type: ignore
         )
 
     return out
@@ -173,7 +220,7 @@ def read_text(path: MPath):
     try:
         click.echo(path.read_text())
     except Exception as exc:  # pragma: no cover
-        raise click.ClickException(exc)
+        raise click.ClickException(str(exc))
 
 
 @mpath.command(help="Print contents of file as JSON.")
@@ -183,4 +230,4 @@ def read_json(path: MPath, indent: int = 4):
     try:
         click.echo(json.dumps(path.read_json(), indent=indent))
     except Exception as exc:  # pragma: no cover
-        raise click.ClickException(exc)
+        raise click.ClickException(str(exc))
diff --git a/mapchete/cli/options.py b/mapchete/cli/options.py
index 8ef7c5f..72493cf 100644
--- a/mapchete/cli/options.py
+++ b/mapchete/cli/options.py
@@ -110,6 +110,20 @@ def _cb_none_concurrency(ctx, param, value):
     return None if value == "none" else value
 
 
+def _cb_mpath(ctx, param, value) -> MPath:
+    if param.name in ("path", "tiledir", "src_tiledir"):
+        return MPath(
+            value,
+            storage_options=ctx.params.get("src_fs_opts"),
+        )
+    elif param.name in ("out_path", "dst_tiledir", "output"):
+        return MPath(
+            value,
+            storage_options=ctx.params.get("dst_fs_opts"),
+        )
+    return MPath(value)  # pragma: no cover
+
+
 # click arguments #
 ###################
 arg_mapchete_file = click.argument("mapchete_file", type=click.Path())
@@ -127,12 +141,12 @@ arg_out_format = click.argument(
 arg_input_raster = click.argument("input_raster", type=click.Path())
 arg_out_dir = click.argument("output_dir", type=click.Path())
 arg_input = click.argument("input_", metavar="INPUT", type=click.STRING)
-arg_output = click.argument("output", type=click.STRING)
-arg_src_tiledir = click.argument("src_tiledir", type=click.STRING)
-arg_dst_tiledir = click.argument("dst_tiledir", type=click.STRING)
-arg_tiledir = click.argument("tiledir", type=click.STRING)
-arg_path = click.argument("path", type=click.Path(path_type=MPath))
-arg_out_path = click.argument("out_path", type=click.Path(path_type=MPath))
+arg_output = click.argument("output", type=click.STRING, callback=_cb_mpath)
+arg_src_tiledir = click.argument("src_tiledir", type=click.STRING, callback=_cb_mpath)
+arg_dst_tiledir = click.argument("dst_tiledir", type=click.STRING, callback=_cb_mpath)
+arg_tiledir = click.argument("tiledir", type=click.STRING, callback=_cb_mpath)
+arg_path = click.argument("path", type=click.STRING, callback=_cb_mpath)
+arg_out_path = click.argument("out_path", type=click.STRING, callback=_cb_mpath)
 
 
 # click options #
diff --git a/mapchete/commands/cp.py b/mapchete/commands/cp.py
index 1c88900..7eb1e5b 100644
--- a/mapchete/commands/cp.py
+++ b/mapchete/commands/cp.py
@@ -49,6 +49,7 @@ def cp(
 
     src_tiledir = MPath.from_inp(src_tiledir, storage_options=src_fs_opts)
     dst_tiledir = MPath.from_inp(dst_tiledir, storage_options=dst_fs_opts)
+
     all_observers = Observers(observers)
 
     # open source tile directory
diff --git a/mapchete/io/_misc.py b/mapchete/io/_misc.py
index 2a0475c..1ef0d7f 100644
--- a/mapchete/io/_misc.py
+++ b/mapchete/io/_misc.py
@@ -1,13 +1,16 @@
 import logging
 from enum import Enum
 
+from fsspec import FSTimeoutError
 import rasterio
 from rasterio.warp import calculate_default_transform
+from retry import retry
 from shapely.errors import TopologicalError
 from shapely.geometry import box
 
 from mapchete.geometry import reproject_geometry, segmentize_geometry
 from mapchete.path import MPath
+from mapchete.settings import IORetrySettings
 from mapchete.tile import BufferedTile, BufferedTilePyramid
 
 logger = logging.getLogger(__name__)
@@ -185,13 +188,30 @@ def get_boto3_bucket(bucket_name):  # pragma: no cover
     raise DeprecationWarning("get_boto3_bucket() is deprecated")
 
 
-def copy(src_path, dst_path, src_fs=None, dst_fs=None, overwrite=False):
+@retry(
+    logger=logger,
+    **dict(IORetrySettings(exceptions=(OSError, FSTimeoutError, TimeoutError))),
+)
+def copy(
+    src_path,
+    dst_path,
+    src_fs=None,
+    dst_fs=None,
+    overwrite=False,
+    exists_ok: bool = False,
+):
     """Copy path from one place to the other."""
     src_path = MPath.from_inp(src_path, fs=src_fs)
     dst_path = MPath.from_inp(dst_path, fs=dst_fs)
 
-    if not overwrite and dst_path.fs.exists(dst_path):
-        raise IOError(f"{dst_path} already exists")
+    if dst_path.fs.exists(dst_path):
+        if overwrite:
+            pass
+        elif exists_ok:
+            logger.debug("%s already exists", str(dst_path))
+            return
+        else:
+            raise IOError(f"{dst_path} already exists")
 
     # create parent directories on local filesystems
     dst_path.parent.makedirs()


Test patch:
diff --git a/test/cli/test_mpath.py b/test/cli/test_mpath.py
index 88d77d4..3ac0a7d 100644
--- a/test/cli/test_mpath.py
+++ b/test/cli/test_mpath.py
@@ -35,6 +35,12 @@ def test_cp(metadata_json, mp_tmpdir):
     assert run_cli(["cp", str(metadata_json), out_file], cli=mpath)
 
 
+def test_cp_directory(local_tiledir, mp_tmpdir):
+    out_path = mp_tmpdir / local_tiledir.name
+    assert run_cli(["cp", str(local_tiledir), str(out_path), "--recursive"], cli=mpath)
+    assert out_path.ls()
+
+
 def test_rm(metadata_json, mp_tmpdir):
     out_file = mp_tmpdir / metadata_json.name
     assert run_cli(["cp", str(metadata_json), out_file], cli=mpath)
diff --git a/test/conftest.py b/test/conftest.py
index 9922d9f..a830242 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -204,6 +204,12 @@ def secure_http_raster(secure_http_testdata_dir):
     return secure_http_testdata_dir / "cleantopo/1/0/0.tif"
 
 
+@pytest.fixture
+def local_tiledir():
+    """Fixture for a local TileDirectory."""
+    return TESTDATA_DIR / "cleantopo"
+
+
 @pytest.fixture
 def http_tiledir(http_testdata_dir):
     """Fixture for HTTP TileDirectory."""
diff --git a/test/test_io.py b/test/test_io.py
index 811f10d..eea7d77 100644
--- a/test/test_io.py
+++ b/test/test_io.py
@@ -332,6 +332,11 @@ def test_copy(cleantopo_br_tif, tmpdir):
     with rasterio_open(out) as src:
         assert not src.read(masked=True).mask.all()
 
+    # copy again but skip if file exists
+    copy(cleantopo_br_tif, out, exists_ok=True)
+    with rasterio_open(out) as src:
+        assert not src.read(masked=True).mask.all()
+
 
 def test_custom_grid_points(custom_grid_points):
     mp = custom_grid_points.process_mp(tile=(3, 1245, 37))