diff data_schema/__init__.py @ 28:db3491e1b590

Allow to customize the loading of the schema dict
author Franz Glasner <fzglas.hg@dom66.de>
date Sat, 08 Jul 2023 13:51:20 +0200
parents fd2a40c3d87b
children 68286d27f27d
line wrap: on
line diff
--- a/data_schema/__init__.py	Sat Jul 08 12:26:44 2023 +0200
+++ b/data_schema/__init__.py	Sat Jul 08 13:51:20 2023 +0200
@@ -37,7 +37,10 @@
 
 import rfc3986
 
-import configmix.yaml
+try:
+    from configmix.yaml import load as default_schema_loader
+except ImportError:
+    default_schema_loader = None
 
 from .util import get_data_stream
 
@@ -233,7 +236,7 @@
 
 ValidationSettings = collections.namedtuple(
     "ValidationSettings",
-    ["skip_keys", "break_on_keynames_problems"])
+    ["skip_keys", "break_on_keynames_problems", "schema_loader"])
 
 
 class _Schema(dict):
@@ -325,13 +328,15 @@
     def __repr__(self):
         return "<_Schema " + super().__repr__() + ">"
 
-    def get_cached_schema(self, key, load_if_needed=True):
+    def get_cached_schema(self, key, load_if_needed=True, schema_loader=None):
         root = self.ROOT
         s = root._schema_cache.get(key, None)
         if s is None and load_if_needed:
+            if schema_loader is None:
+                raise SchemaError("no schema loader available")
             with get_data_stream(key) as schemastream:
                 # load schema a new `$self' (i.e. sub-root is True)
-                s = _Schema(self, True, configmix.yaml.load(schemastream))
+                s = _Schema(self, True, schema_loader(schemastream))
             root._schema_cache[key] = s
         return s
 
@@ -496,6 +501,7 @@
     settings = {
         "skip_keys": None,
         "break_on_keynames_problems": True,
+        "schema_loader": default_schema_loader
     }
     settings.update(kwds)
     if not isinstance(schema, _Schema):
@@ -1399,7 +1405,10 @@
         elif uri.path == SCHEMA_PATH_ROOT:
             s = schema.ROOT
         else:
-            s = schema.get_cached_schema(uri.path, load_if_needed=True)
+            s = schema.get_cached_schema(
+                uri.path,
+                load_if_needed=True,
+                schema_loader=context.settings.schema_loader)
         if uri.fragment is None:
             raise SchemaError("fragment required in reference")