feat: add custom validation extension

src/simdb/cli/commands/simulation.py

@@ -1,13 +1,13 @@
-import click
 from pathlib import Path
-from typing import Optional, List, Tuple, Any, Type
+from typing import Any, List, Optional, Tuple, Type
+
+import click
 
-from . import pass_config, check_meta_args
 from ...config.config import Config
 from ...query import QueryType, parse_query_arg
+from . import check_meta_args, pass_config
 from .validators import validate_non_negative
 
-
 # def _validate_simulation_outputs(options: dict, simulation):
 #     file_validator_type = options.get("file_validator", None)
 #     file_validator_options = options.get("file_validator_options", {})
@@ -170,9 +170,10 @@ def simulation_info(config: Config, sim_id: str):
 def simulation_ingest(config: Config, manifest_file: str, alias: str):
     """Ingest a MANIFEST_FILE."""
     import urllib.parse
+
     from ...database import get_local_db
     from ...database.models import Simulation
-    from ..manifest import Manifest, InvalidAlias
+    from ..manifest import InvalidAlias, Manifest
 
     manifest = Manifest()
     manifest.load(Path(manifest_file))
@@ -233,10 +234,11 @@ def simulation_push(
     add_watcher: bool,
 ):
     """Push the simulation with the given SIM_ID (UUID or alias) to the REMOTE."""
+    import sys
+
     from ...database import get_local_db
+    from ...validation import ValidationError, Validator
     from ..remote_api import RemoteAPI
-    from ...validation import Validator, ValidationError
-    import sys
 
     api = RemoteAPI(remote, username, password, config)
     db = get_local_db(config)
@@ -251,7 +253,7 @@ def simulation_push(
     schemas = api.get_validation_schemas()
     try:
         for schema in schemas:
-            Validator(schema).validate(simulation)
+            Validator(schema, config).validate(simulation)
     except ValidationError as err:
         raise click.ClickException(f"Simulation does not validate: {err}")
 
@@ -279,10 +281,11 @@ def simulation_pull(
     password: Optional[str],
 ):
     """Pull the simulation with the given SIM_ID (UUID or alias) from the REMOTE."""
-    from ...database import get_local_db, DatabaseError
-    from ..remote_api import RemoteAPI, RemoteError
     import sys
 
+    from ...database import DatabaseError, get_local_db
+    from ..remote_api import RemoteAPI, RemoteError
+
     api = RemoteAPI(remote, username, password, config)
     db = get_local_db(config)
 
@@ -323,7 +326,9 @@ def simulation_pull(
     help="Include UUID in the output.",
     default=False,
 )
-def simulation_query(config: Config, constraints: List[str], meta: List[str], show_uuid: bool):
+def simulation_query(
+    config: Config, constraints: List[str], meta: List[str], show_uuid: bool
+):
     """Perform a metadata query to find matching local simulations.
 
     \b
@@ -381,7 +386,9 @@ def simulation_query(config: Config, constraints: List[str], meta: List[str], sh
 
     db = get_local_db(config)
     simulations = db.query_meta(parsed_constraints)
-    print_simulations(simulations, verbose=config.verbose, metadata_names=names, show_uuid=show_uuid)
+    print_simulations(
+        simulations, verbose=config.verbose, metadata_names=names, show_uuid=show_uuid
+    )
 
 
 @simulation.command("validate", cls=n_required_args_adaptor(1))
@@ -395,6 +402,7 @@ def simulation_validate(
 ):
     """Validate the ingested simulation with given SIM_ID (UUID or alias) using validation schema from REMOTE."""
     from itertools import chain
+
     from ...database import get_local_db
     from ...validation import ValidationError, Validator
     from ..remote_api import RemoteAPI
@@ -410,7 +418,7 @@ def simulation_validate(
 
     click.echo("validating metadata ... ", nl=False)
     for schema in schemas:
-        Validator(schema).validate(simulation)
+        Validator(schema, config).validate(simulation)
 
     ids_list = []
     for file in chain(simulation.inputs, simulation.outputs):

src/simdb/remote/apis/v1/simulations.py

@@ -12,12 +12,12 @@
 from ....database.models import simulation as models_sim
 from ....uri import URI
 from ... import APIConstants
-from ...core.typing import current_app
 from ...core.alias import create_alias_dir
 from ...core.auth import User, requires_auth
 from ...core.cache import cache, cache_key, clear_cache
 from ...core.errors import error
 from ...core.path import secure_path
+from ...core.typing import current_app
 
 api = Namespace("simulations", path="/")
 
@@ -52,7 +52,7 @@ def _validate(simulation, user) -> Dict:
 
     schema = Validator.validation_schema()
     try:
-        Validator(schema).validate(simulation)
+        Validator(schema, current_app.simdb_config).validate(simulation)
         _update_simulation_status(simulation, models_sim.Simulation.Status.PASSED, user)
         return {
             "passed": True,

src/simdb/remote/apis/v1_1/simulations.py

@@ -12,12 +12,12 @@
 from ....database.models import simulation as models_sim
 from ....uri import URI
 from ... import APIConstants
-from ...core.typing import current_app
 from ...core.alias import create_alias_dir
 from ...core.auth import User, requires_auth
 from ...core.cache import cache, cache_key, clear_cache
 from ...core.errors import error
 from ...core.path import secure_path
+from ...core.typing import current_app
 
 api = Namespace("simulations", path="/")
 
@@ -53,7 +53,7 @@ def _validate(simulation, user) -> Dict:
     schemas = Validator.validation_schemas(current_app.simdb_config, simulation)
     try:
         for schema in schemas:
-            Validator(schema).validate(simulation)
+            Validator(schema, current_app.simdb_config).validate(simulation)
             _update_simulation_status(
                 simulation, models_sim.Simulation.Status.PASSED, user
             )

src/simdb/remote/apis/v1_2/simulations.py

@@ -12,13 +12,12 @@
 from ....database.models import simulation as models_sim
 from ....database.models import watcher as models_watcher
 from ....uri import URI
-from ....cli.manifest import DataObject
-from ...core.typing import current_app
 from ...core.alias import create_alias_dir
 from ...core.auth import User, requires_auth
 from ...core.cache import cache, cache_key, clear_cache
 from ...core.errors import error
-from ...core.path import secure_path, find_common_root
+from ...core.path import find_common_root, secure_path
+from ...core.typing import current_app
 
 api = Namespace("simulations", path="/")
 
@@ -57,7 +56,7 @@ def _validate(simulation, user) -> Dict:
     schemas = Validator.validation_schemas(current_app.simdb_config, simulation)
     try:
         for schema in schemas:
-            Validator(schema).validate(simulation)
+            Validator(schema, current_app.simdb_config).validate(simulation)
             _update_simulation_status(
                 simulation, models_sim.Simulation.Status.PASSED, user
             )
@@ -72,7 +71,6 @@ def _validate(simulation, user) -> Dict:
     file_validator_options = current_app.simdb_config.get_section("file_validation", default={})
     if file_validator_type not in [None, "none",""]:
         from ....validation.file import find_file_validator
-        from imas_validator.validate_options import ValidateOptions
         validator_type, validator_options  = find_file_validator(file_validator_type, file_validator_options)
         if validator_type:
 
@@ -154,9 +152,10 @@ def _get_json_aware(force: bool = False, silent: bool = False):
     - force/silent mimic request.get_json behavior.
     - Uses Flask's JSON provider to ensure identical types/decoding.
     """
-    from flask import current_app
     import gzip
 
+    from flask import current_app
+
     # Match request.get_json content-type check unless forced
     if not force:
         mimetype = (request.mimetype or "")

src/simdb/validation/validator.py

@@ -1,11 +1,13 @@
-import cerberus
-import yaml
 import re
+from importlib import import_module
 from pathlib import Path
 from typing import Dict, List, Optional
 
-from ..database.models.simulation import Simulation
+import cerberus
+import yaml
+
 from ..config import Config, ConfigError
+from ..database.models.simulation import Simulation
 
 
 class TestParameters:
@@ -69,9 +71,10 @@ def _validate_max_value(self, max_value, field, value):
             self._error(field, "Maximum %s greater than %s" % (value.max(), max_value))
 
     def _compare(self, comparison, field, value, comparator: str, message: str):
-        import numpy as np        
+        import numpy as np
+
         if comparison is None:
-            return        
+            return
         if isinstance(value, np.ndarray):
             value = value[~np.isnan(value)]
             if value.size == 0:
@@ -190,9 +193,23 @@ def validation_schemas(
 
         return schemas
 
-    def __init__(self, schema: Dict):
+    def _validation_extension(self, config: Config):
+        schema_path = config.get_option("validation.custom_validator", default=None)
+        if schema_path is None:
+            return CustomValidator
+
+        module_name, class_name = schema_path.rsplit(".", 1)
+        module = import_module(module_name)
+        validation_cls = getattr(module, class_name)
+
+        if not issubclass(validation_cls, CustomValidator):
+            raise (TypeError(f"{validation_cls.__name__} must inherit CustomValidator"))
+        return validation_cls
+
+    def __init__(self, schema: Dict, config: Config):
         try:
-            self._validator = CustomValidator(schema)
+            validation_cls = self._validation_extension(config)
+            self._validator = validation_cls(schema)
             self._validator.allow_unknown = True
         except cerberus.SchemaError:
             raise LoadError("Failed to parse validation schema")