feat: add named schemas

ckanext/transmute/interfaces.py

@@ -1,3 +1,6 @@
+from __future__ import annotations
+
+from typing import Any
 from ckan.plugins.interfaces import Interface
 
 
@@ -18,3 +21,12 @@ def get_transmutators(self):
         These transmutator functions would then be available for
         tsm_schema.
         """
+
+    def get_transmutation_schemas(self) -> dict[str, dict[str, Any]]:
+        """Return definitions of named schemas.
+
+        These schemas can be reffered by name in code. In this way you can
+        define static schema and apply in multiple places it to arbitrary data.
+        """
+
+        return {}

ckanext/transmute/plugin.py

@@ -1,3 +1,7 @@
+from __future__ import annotations
+
+import json
+from typing import Any
 import ckan.plugins as p
 import ckan.plugins.toolkit as tk
 
@@ -6,6 +10,8 @@
 from ckanext.transmute.transmutators import get_transmutators
 from ckanext.transmute.interfaces import ITransmute
 
+from . import utils
+
 
 class TransmutePlugin(p.SingletonPlugin):
     p.implements(p.IConfigurer)
@@ -17,6 +23,7 @@ class TransmutePlugin(p.SingletonPlugin):
     def update_config(self, config_):
         tk.add_template_directory(config_, "templates")
         tk.add_resource("assets", "transmute")
+        utils.collect_schemas()
 
     # IActions
     def get_actions(self):
@@ -31,3 +38,11 @@ def get_auth_functions(self):
     # ITransmute
     def get_transmutators(self):
         return get_transmutators()
+
+    def get_transmutation_schemas(self) -> dict[str, Any]:
+        prefix = "ckanext.transmute.schema."
+        return {
+            key[len(prefix) :]: json.load(open(tk.config[key]))
+            for key in tk.config
+            if key.startswith(prefix)
+        }

ckanext/transmute/schema.py

@@ -98,7 +98,7 @@ def get_root_type(self):
         if not root_type:
             raise SchemaParsingError("Schema: root type is missing")
 
-        if not root_type in self.schema.get("types"):
+        if not root_type in self.schema.get("types", []):
             raise SchemaParsingError("Schema: root_type is declared but not defined")
 
         return root_type

ckanext/transmute/utils.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
+
 import logging
+from typing import Any, Callable
 
 import ckan.plugins as p
 
@@ -9,10 +11,23 @@
 
 SENTINEL = {}
 _transmutator_cache = {}
+_schema_cache = {}
+
 log = logging.getLogger(__name__)
 
 
-def get_transmutator(transmutator: str):
+def get_schema(name: str) -> dict[str, Any] | None:
+    """Return named schema."""
+    return _schema_cache.get(name)
+
+
+def collect_schemas():
+    """Collect named schemas from ITransmute plugins."""
+    for plugin in reversed(list(p.PluginImplementations(ITransmute))):
+        _schema_cache.update(plugin.get_transmutation_schemas())
+
+
+def get_transmutator(transmutator: str) -> Callable[..., Any]:
     get_all_transmutators()
 
     try:
@@ -33,7 +48,7 @@ def get_all_transmutators() -> list[str]:
     return list(_transmutator_cache.keys())
 
 
-def get_json_schema():
+def get_json_schema() -> dict[str, Any]:
     transmutators = get_all_transmutators()
     return {
         "$schema": "http://json-schema.org/draft-04/schema",
@@ -116,7 +131,10 @@ def get_json_schema():
                                                     "oneOf": [
                                                         {
                                                             "type": "string",
-                                                            "enum": [MODE_COMBINE, MODE_FIRST_FILLED],
+                                                            "enum": [
+                                                                MODE_COMBINE,
+                                                                MODE_FIRST_FILLED,
+                                                            ],
                                                         }
                                                     ]
                                                 },