validator package does not assume a odoo.addons package name, provide full package name instead

validate.py

@@ -18,4 +18,5 @@
 #
 ##############################################################################
 import json
+import logging
 import os
@@ -21,2 +22,3 @@
 import os
+from collections.abc import Callable
 from enum import Enum
@@ -22,4 +24,5 @@
 from enum import Enum
+from importlib import import_module
 from typing import Any, LiteralString
 
 import fastjsonschema  # type: ignore[import-untyped]
@@ -23,7 +26,9 @@
 from typing import Any, LiteralString
 
 import fastjsonschema  # type: ignore[import-untyped]
-import odoo.addons  # type: ignore[import-untyped]
+from odoo.exceptions import UserError  # type: ignore[import-untyped]
+
+_logger = logging.getLogger(__name__)
 
 
 class Validation(str, Enum):
@@ -38,6 +43,13 @@
     pass
 
 
+def _add_schema(schemas, schema):
+    if "$id" in schema:
+        schemas[schema["$id"]] = schema
+    else:
+        _logger.warning("Schema without $id (schema ignored)")
+
+
 class Validator:
     def __init__(
         self,
@@ -41,6 +53,5 @@
 class Validator:
     def __init__(
         self,
-        repository_module_name: str,
-        repository: str,
+        package_name: str,
         default_url_pattern: str,
@@ -46,2 +57,3 @@
         default_url_pattern: str,
+        directory: str | None = None,
     ):
@@ -47,5 +59,10 @@
     ):
-        self.repository_module_name = repository_module_name
-        self.repository = repository
+        """
+        :param package_name: Package where the schema can be found
+        :param default_url_pattern: pattern for url ({} will be replaced by $id)
+        :param directory: directory to search for json, not used if a get_schema is
+        provided in the package.
+        """
+        self.package_name = package_name
         # exemple "https://annonces-legales.fr/xbus/schemas/v1/{}.schema.json"
         self.default_url_pattern = default_url_pattern
@@ -50,5 +67,5 @@
         # exemple "https://annonces-legales.fr/xbus/schemas/v1/{}.schema.json"
         self.default_url_pattern = default_url_pattern
-        self.validators: dict[LiteralString, Any] = {}
+        self.validators: dict[LiteralString, Callable] = {}
         self.initialized = False
         self.encoding = "UTF-8"
@@ -53,4 +70,5 @@
         self.initialized = False
         self.encoding = "UTF-8"
+        self.directory = directory
 
     def initialize(self) -> None:
@@ -55,23 +73,30 @@
 
     def initialize(self) -> None:
-        # TODO Not working if module is installed compressed
-        repo_module_basepath = os.path.dirname(
-            getattr(odoo.addons, self.repository_module_name).__file__
-        )
-
-        # Read local schema definitions.
-        schemas = {}
-        schema_search_path = os.path.abspath(
-            os.path.join(repo_module_basepath, self.repository)
-        )
-        for root, _dirs, files in os.walk(schema_search_path):
-            for fname in files:
-                fpath = os.path.join(root, fname)
-                if fpath.endswith((".json",)):
-                    with open(fpath, encoding=self.encoding) as schema_fd:
-                        schema = json.load(schema_fd)
-                        if "$id" in schema:
-                            schemas[schema["$id"]] = schema
+        if self.initialized:
+            return
+        schemas: dict[LiteralString, Any] = {}
+        module = import_module(self.package_name)
+        if hasattr(module, "get_schemas"):
+            for schema in module.get_schemas():
+                _add_schema(schemas, schema)
+        else:
+            if module.__file__ is None:
+                # XXX maybe not the best type of error
+                raise UserError("Module %s has no file", self.package_name)
+            # Fallback on searching schema json files
+            schema_search_path = os.path.dirname(module.__file__)
+            schema_search_path = os.path.abspath(
+                os.path.join(schema_search_path, self.directory)
+                if self.directory is not None
+                else schema_search_path
+            )
+            for root, _dirs, files in os.walk(schema_search_path):
+                for fname in files:
+                    fpath = os.path.join(root, fname)
+                    if fpath.endswith((".json",)):
+                        with open(fpath, encoding=self.encoding) as schema_fd:
+                            schema = json.load(schema_fd)
+                            _add_schema(schemas, schema)
 
         # Prepare validators for each schema. We add an HTTPS handler that
         # points back to our schema definition cache built above.