Linting and easier version handling

Author: m-mohr

tests/test_create_geoparquet.py

@@ -4,6 +4,7 @@
 
 from vecorel_cli.create_geoparquet import CreateGeoParquet
 from vecorel_cli.encoding.geoparquet import GeoParquet
+from vecorel_cli.vecorel.schemas import Schemas
 
 
 def test_create_geoparquet(tmp_parquet_file: Path):
@@ -22,6 +23,7 @@ def test_create_geoparquet(tmp_parquet_file: Path):
     schema_ids = collection["schemas"].keys()
     assert len(schema_ids) == 1
     assert "inspire" in schema_ids
+    assert all(Schemas.get_core_uri() in s for s in collection["schemas"].values())
 
     data = gp.read()
     assert len(data) == 2

tests/test_jsonschema.py

@@ -2,11 +2,10 @@
 from vecorel_cli.encoding.geojson import GeoJSON
 from vecorel_cli.validate_schema import ValidateSchema
 from vecorel_cli.vecorel.util import load_file
-from vecorel_cli.vecorel.version import vecorel_version
 
 vecorel_schema_uri = "tests/data-files/jsonschema/schema.yaml"
 vecorel_schema = load_file(vecorel_schema_uri)
-datatypes_uri = GeoJSON.get_datatypes_uri(vecorel_version)
+datatypes_uri = GeoJSON.get_datatypes_uri()
 datatypes = GeoJSON.load_datatypes(datatypes_uri)
 schema_id = "https://example.com/schema.json"
 point = {"type": "Point", "coordinates": [0, 0]}

vecorel_cli/cli/logger.py

@@ -1,7 +1,7 @@
 import re
 import sys
 from logging import Logger
-from typing import Any, Optional
+from typing import Any, Optional, Union
 
 from loguru import logger
 
@@ -36,7 +36,7 @@ def info(self, message: str, **kwargs):
     def warning(self, message: str, **kwargs):
         self.log(message, "warning", **kwargs)
 
-    def error(self, message: str, **kwargs):
+    def error(self, message: Union[Exception, str], **kwargs):
         self.log(message, "error", **kwargs)
 
     def success(self, message: str, **kwargs):
@@ -58,7 +58,9 @@ def log(
         # Escape XML/HTML tags etc. that look like loguru color directives
         # The regexp is coming directly from the loguru source code, see
         # https://github.com/Delgan/loguru/blob/master/loguru/_colorizer.py
-        message = re.sub(r"(</?(?:[fb]g\s)?[^<>\s]*>)", r"\\\1", message, count=0, flags=re.IGNORECASE)
+        message = re.sub(
+            r"(</?(?:[fb]g\s)?[^<>\s]*>)", r"\\\1", message, count=0, flags=re.IGNORECASE
+        )
 
         # Handle indentation (including multiple lines)
         message = self._indent_text(message, indent)

vecorel_cli/conversion/base.py

@@ -30,7 +30,6 @@
 from ..vecorel.schemas import Schemas
 from ..vecorel.typing import Sources
 from ..vecorel.util import get_fs, name_from_uri, stream_file
-from ..vecorel.version import vecorel_version
 
 
 class BaseConverter(LoggerMixin):
@@ -274,7 +273,7 @@ def get_title(self):
         return f"{title} ({self.variant})" if self.variant else title
 
     def create_collection(self, cid) -> Collection:
-        schema_uris = [Schemas.get_core_uri(vecorel_version)]
+        schema_uris = [Schemas.get_core_uri()]
         schema_uris.extend(self.extensions)
         collection = Collection(
             {

vecorel_cli/create_jsonschema.py

@@ -31,7 +31,7 @@ def get_cli_args():
                 type=PathOrURL(extensions=[".json"]),
                 help=f"Vecorel schema to create the JSON Schema for. Can be a local file or a URL. If not provided, loads the schema for Vecorel version {vecorel_version}.",
                 show_default=True,
-                default=Schemas.get_core_uri(vecorel_version),
+                default=Schemas.get_core_uri(),
             ),
             "datatypes": click.option(
                 "--datatypes",
@@ -40,7 +40,7 @@ def get_cli_args():
                 type=PathOrURL(extensions=[".json"]),
                 help=f"Schema for the Vecorel GeoJSON datatypes. Can be a local file or a URL. If not provided, loads the GeoJSON datatypes for Vecorel version {vecorel_version}.",
                 show_default=True,
-                default=GeoJSON.get_datatypes_uri(vecorel_version),
+                default=GeoJSON.get_datatypes_uri(),
             ),
             "target": VECOREL_TARGET_CONSOLE,
             "id": click.option(

vecorel_cli/encoding/base.py

@@ -37,15 +37,16 @@ def get_summary(self) -> dict:
             "Compression": self.get_compression() or "None",
         }
 
-    def _load_collection(self) -> dict:
+    def _load_collection(self) -> Union[Collection, dict]:
         return {}
 
     def get_collection(self) -> Collection:
         """
         Get the collection metadata.
         """
         if self.collection is None:
-            self.collection = Collection(self._load_collection())
+            c = self._load_collection()
+            self.collection = c if isinstance(c, Collection) else Collection(c)
         return self.collection
 
     def set_collection(self, collection: Union[Collection, dict]):

vecorel_cli/encoding/geojson.py

@@ -14,6 +14,7 @@
 from ..vecorel.collection import Collection
 from ..vecorel.typing import Feature, FeatureCollection, SchemaMapping
 from ..vecorel.util import load_file, to_iso8601
+from ..vecorel.version import vecorel_version
 from .base import BaseEncoding
 
 
@@ -34,16 +35,18 @@ def load_datatypes(uri: Union[Path, URL, str]) -> dict:
         return response.get("$defs", {})
 
     @staticmethod
-    def get_datatypes_uri(version: str) -> str:
+    def get_datatypes_uri(version: Optional[str] = None) -> str:
+        if version is None:
+            version = vecorel_version
         return GeoJSON.datatypes_schema_uri.format(version=version)
 
     def get_format(self) -> str:
         return "GeoJSON"
 
-    def _load_collection(self) -> Collection:
+    def _load_collection(self) -> Union[Collection, dict]:
         if self.fs.exists(self.uri):
             self.read(num=0)
-        return self.collection or Collection()
+        return self.collection or {}
 
     def get_validator(self) -> Optional[Validator]:
         from ..validation.geojson import GeoJSONValidator

vecorel_cli/registry.py

@@ -33,9 +33,12 @@ class VecorelRegistry:
         "collection",
     ]
 
-    # Any schemas that are mandatory to have, except for the vecorel specification schema
+    # Any schemas patterns that are mandatory to have, except for the vecorel specification schema
     required_extensions: list[Union[str, re.Pattern]] = []
 
+    # Any specific schemas URIs that are added by default when creating a new dataset
+    default_extensions: list[str] = []
+
     # The filenames for datasets (converters) that should be ignored by the CLI.
     # Always ignores files with a starting "." or "__"
     ignored_datasets: list[str] = [

vecorel_cli/validate.py

@@ -35,14 +35,14 @@ def get_cli_args():
     def validate_cli(
         self,
         source: list[Union[str, Path]],
-        num: int = 100,
+        num: Optional[int] = 100,
         schema_map: SchemaMapping = {},
     ):
         if not isinstance(source, list):
             raise ValueError("Source must be a list.")
         if len(source) == 0:
             raise ValueError("No source files provided")
-        if num < 0:
+        if num is not None and num < 0:
             num = None
 
         invalid = 0
@@ -102,6 +102,8 @@ def validate(
     ) -> Validator:
         encoding = create_encoding(file)
         validator = encoding.get_validator()
+        if validator is None:
+            raise ValueError(f"No validator available for files of type {encoding.get_format()}")
         validator.set_required_schemas(Registry.required_extensions)
         validator.validate(num=num, schema_map=schema_map)
         return validator

vecorel_cli/vecorel/schemas.py

@@ -6,7 +6,7 @@
 
 from ..vecorel.typing import RawSchemas, SchemaMapping
 from ..vecorel.util import load_file
-from .version import is_sdl_supported, sdl_uri
+from .version import is_sdl_supported, sdl_uri, vecorel_version
 
 if TYPE_CHECKING:
     from ..validation.base import Validator
@@ -314,7 +314,9 @@ class Schemas(dict):
     spec_schema = "https://vecorel.org/specification/v{version}/schema.yaml"
 
     @staticmethod
-    def get_core_uri(version: str) -> str:
+    def get_core_uri(version: Optional[str] = None) -> str:
+        if version is None:
+            version = vecorel_version
         return Schemas.spec_schema.format(version=version)
 
     def __init__(self, schemas: Union[RawSchemas, "Schemas"] = {}):