Mercurial > hgrepos > Python > libs > data-schema
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")
