Support loading from remote resource and support loading from folders

Author: m-mohr

pyproject.toml

@@ -20,6 +20,27 @@ classifiers = [
     "Programming Language :: Python :: 3.13",
 ]
 requires-python = ">=3.10"
+# This is duplicate, must also be changed below in the pixi configuration
+dependencies = [
+    "pyyaml>=6.0,<7.0",
+    "click>=8.1,<9.0",
+    "geopandas>=1.0.0,<2.0",
+    "requests>=2.30,<3.0",
+    "shapely>=2.1,<3.0",
+    # numpy is restricted <2.2.0 due to the lower supported Python version being 3.11 and we still cater for 3.10
+    "numpy>=2.0,<2.2",
+    "pyarrow>=21.0,<22.0",
+    "py7zr>=1.0,<2.0",
+    "fsspec==2025.7.0",
+    "jsonschema[format]>=4.20,<5.0",
+    "aiohttp>=3.9,<4.0",
+    "yarl>=1.20,<2.0",
+    "rarfile>=4.0,<5.0",
+    "semantic-version>=2.10.0,<3.0",
+    "json-stream>=2.3.0,<3.0",
+    "flatdict>=4.0,<5.0",
+    "loguru==0.7.3",
+]
 
 [project.scripts]
 vec = "vecorel_cli:vecorel_cli"
@@ -49,15 +70,6 @@ pyarrow = ">=21.0,<22.0"
 py7zr = ">=1.0,<2.0"
 
 [tool.pixi.pypi-dependencies]
-fsspec = "==2025.7.0"
-jsonschema = {version = ">=4.20,<5.0", extras = ["format"]}
-aiohttp = ">=3.9,<4.0"
-rarfile = ">=4.0,<5.0"
-semantic-version = ">=2.10.0,<3.0"
-json-stream = ">=2.3.0,<3.0"
-flatdict = ">=4.0,<5.0"
-loguru = "==0.7.3"
-
 # Editable install of the project itself
 vecorel-cli = {path = ".", editable = true}
 

tests/test_encoding_auto.py

@@ -16,7 +16,6 @@
     # non-existing encoding
     ("invalid.txt", None),
     ("invalid", None),
-    (None, None),
 ]
 
 

tests/test_encoding_geojson.py

@@ -10,8 +10,8 @@ def test_init_paths(tmp_folder):
     fpath = str(tmp_folder / "test.json")
     ppath = Path(fpath)
 
-    assert GeoJSON(fpath).file == ppath
-    assert GeoJSON(ppath).file == ppath
+    assert GeoJSON(fpath).uri == ppath
+    assert GeoJSON(ppath).uri == ppath
 
 
 def test_get_datatypes_uri():

tests/test_encoding_geoparquet.py

@@ -8,8 +8,8 @@ def test_init_paths(tmp_folder):
     fpath = str(tmp_folder / "test.parquet")
     ppath = Path(fpath)
 
-    assert GeoParquet(fpath).file == ppath
-    assert GeoParquet(ppath).file == ppath
+    assert GeoParquet(fpath).uri == ppath
+    assert GeoParquet(ppath).uri == ppath
 
 
 def test_get_format():

vecorel_cli/basecommand.py

@@ -39,6 +39,7 @@ def run(self, *args, **kwargs):
             f"{Registry.cli_title} {Registry.get_version()} - {self.cmd_title}",
             end="\n\n",
             style="bold",
+            color="cyan",
         )
 
         # Detect method to run

vecorel_cli/cli/logger.py

@@ -100,7 +100,7 @@ def _format_data(self, value: dict, depth=0, max_depth=1, strlen=50):
                 if depth > 0:
                     output += "\n"
                 for key, value in value.items():
-                    output += f"{prefix}<yellow>{key}</>: "
+                    output += f"{prefix}<cyan>{key}</>: "
                     output += self._format_data(
                         value, depth=depth + 1, max_depth=max_depth, strlen=strlen
                     )

vecorel_cli/cli/options.py

@@ -1,7 +1,9 @@
 import click
 
 from ..const import COMPRESSION_METHODS, GEOPARQUET_DEFAULT_VERSION, GEOPARQUET_VERSIONS
-from .util import valid_schemas_for_cli, valid_vecorel_file, valid_vecorel_files
+from ..registry import Registry
+from .path_url import PathOrURL
+from .util import valid_schemas_for_cli
 
 
 def CRS(default_value):
@@ -62,16 +64,15 @@ def CRS(default_value):
 
 VECOREL_FILES_ARG = click.argument(
     "source",
-    type=click.Path(exists=True, dir_okay=False, resolve_path=True),
+    type=PathOrURL(multiple=True, extensions=Registry.get_vecorel_extensions()),
     nargs=-1,
-    callback=valid_vecorel_files,
+    callback=PathOrURL.flatten_tuples,
 )
 
 VECOREL_FILE_ARG = click.argument(
     "source",
-    type=click.Path(exists=True, dir_okay=False, resolve_path=True),
+    type=PathOrURL(extensions=Registry.get_vecorel_extensions()),
     nargs=1,
-    callback=valid_vecorel_file,
 )
 
 

vecorel_cli/cli/path_url.py

@@ -0,0 +1,70 @@
+import click
+import pathlib
+
+from ..vecorel.util import is_url, get_fs
+from yarl import URL
+
+IGNORE_FILES = ["collection.json", "catalog.json"] # likely STAC
+
+class PathOrURL(click.ParamType):
+    name = "path_or_url"
+
+    @staticmethod
+    def flatten_tuples(ctx, param, value) -> list:
+        if not value:
+            return []
+        data = []
+        for v in value:
+            if isinstance(v, tuple) or isinstance(v, list):
+                data.extend(v)
+            else:
+                data.append(v)
+        return data
+
+    def __init__(self, *, multiple: bool = False, extensions: list[str] = []):
+        self.extensions = extensions
+        self.multiple = multiple
+        self.path_type = click.Path(
+            exists=True,
+            dir_okay=multiple,
+            resolve_path=True,
+            allow_dash=False,
+            path_type=pathlib.Path,
+        )
+
+    def convert(self, value, param, ctx):
+        # Check if it's a URL
+        if "://" in value and is_url(value):
+            fs = get_fs(value)
+            if fs.exists(value):
+                return URL(value)
+            else:
+                self.fail(f"URL '{value}' is does not exist or is currently unavailable.", param, ctx)
+
+        # Otherwise, validate as a local path
+        filepath = self.path_type.convert(value, param, ctx)
+        if filepath.is_dir():
+            files = []
+            for f in filepath.iterdir():
+                if not self._check_extension(f):
+                    continue
+                if f.name in IGNORE_FILES:
+                    continue
+
+                files.append(f)
+            return tuple(files)
+        elif not self._check_extension(filepath):
+            self.fail(
+                f"File '{filepath}' must have one of the following extensions: {', '.join(self.extensions)}",
+                param,
+                ctx
+            )
+        return filepath
+
+    def _check_extension(self, filepath: pathlib.Path) -> bool:
+        return len(self.extensions) == 0 or filepath.suffix.lower() in self.extensions
+
+    def shell_complete(self, ctx, param, incomplete):
+        if "://" in incomplete:
+            return []
+        return super().shell_complete(ctx, param, incomplete)

vecorel_cli/cli/util.py

@@ -1,90 +1,10 @@
-import os
 from pathlib import Path
 from typing import Optional
-from urllib.parse import urlparse
 
 import click
 import pandas as pd
 
-from ..const import SUPPORTED_PROTOCOLS
-from ..registry import Registry
-from ..vecorel.util import name_from_uri
-
-
-def is_valid_file_uri(uri, extensions=[]):
-    """Determine if the input is a file path or a URL and handle it."""
-    if not isinstance(uri, str):
-        raise click.BadParameter("Input must be a string representing a file path or URL")
-    elif len(extensions) > 0 and not uri.endswith(tuple(extensions)):
-        raise click.BadParameter(
-            f"File '{uri}' must have one of the following extensions: {', '.join(extensions)}"
-        )
-    elif os.path.exists(uri):
-        return uri
-    elif is_valid_url(uri):
-        return uri
-    else:
-        raise click.BadParameter(
-            "Input must be an existing local file or a URL with protocol: "
-            + ",".join(SUPPORTED_PROTOCOLS)
-        )
-
-
-def is_valid_url(url):
-    """Check if a URL is valid."""
-    try:
-        result = urlparse(url)
-        return all([result.scheme in SUPPORTED_PROTOCOLS, result.netloc])
-    except ValueError:
-        return False
-
-
-def get_files(value, extensions=[]):
-    files = []
-    extensions = tuple(extensions)
-    for v in value:
-        v = is_valid_file_uri(v)
-        if os.path.isdir(v):
-            for f in os.listdir(v):
-                if len(extensions) > 0 and not f.endswith(extensions):
-                    continue
-                if f == "collection.json" or f == "catalog.json":  # likely STAC
-                    continue
-                files.append(os.path.join(v, f))
-        else:
-            files.append(v)
-    return files
-
-
-def valid_file(ctx, param, value):
-    return is_valid_file_uri(value)
-
-
-def valid_vecorel_file(ctx, param, value) -> Path:
-    ext = Registry.get_vecorel_extensions()
-    return is_valid_file_uri(value, extensions=ext)
-
-
-def valid_vecorel_files(ctx, param, value) -> list[Path]:
-    ext = Registry.get_vecorel_extensions()
-
-    files = []
-    if isinstance(value, str):
-        files = [value]
-    elif isinstance(value, tuple):
-        files = list(value)
-    elif isinstance(value, list):
-        files = value
-
-    if len(files) == 0:
-        raise click.BadParameter("No files provided.")
-
-    actual_files = []
-    for file in files:
-        if is_valid_file_uri(file, extensions=ext):
-            actual_files.append(Path(file))
-
-    return actual_files
+from ..vecorel.util import is_url, name_from_uri
 
 
 def parse_converter_input_files(ctx, param, value):
@@ -130,14 +50,14 @@ def valid_schemas_for_cli(value: tuple[str]) -> dict[str, Path]:
 
         if len(part) != 2:
             raise click.BadParameter(
-                "Schema must be a URL and a local file path separated by a comma character."
+                "Schema must be a URL and a local file path, separated by a comma character."
             )
-        if not is_valid_url(part[0]):
+        if not is_url(part[0]):
             raise click.BadParameter(f"Schema URL '{part[0]}' is not a valid URL.")
 
         p = Path(part[1])
         if not p.exists():
-            raise click.BadParameter(f"Local schema file '{p}' does not exist.")
+            raise click.BadParameter(f"Local schema file '{p.resolve()}' does not exist.")
 
         map_[part[0]] = p
 

vecorel_cli/create_geojson.py

@@ -2,6 +2,7 @@
 from typing import Optional, Union
 
 import click
+from yarl import URL
 
 from .basecommand import BaseCommand, runnable
 from .cli.options import JSON_INDENT, PROPERTIES, VECOREL_FILE_ARG, VECOREL_TARGET
@@ -43,7 +44,7 @@ def get_cli_args():
     @runnable
     def create(
         self,
-        source: Union[Path, str],
+        source: Union[Path, URL, str],
         target: Union[Path, str],
         properties: Optional[Union[tuple[str], list[str]]] = None,
         split: bool = False,