Mercurial > hgrepos > Python > libs > data-schema
changeset 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 | 88ee7d1cc0bb |
| children | 68286d27f27d |
| files | data_schema/__init__.py docs/schema.txt tests/test_schema.py |
| diffstat | 3 files changed, 75 insertions(+), 18 deletions(-) [+] |
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")
--- a/docs/schema.txt Sat Jul 08 12:26:44 2023 +0200 +++ b/docs/schema.txt Sat Jul 08 13:51:20 2023 +0200 @@ -45,6 +45,13 @@ Default: ``True`` +- ``schema_loader`` + + Default: configmix.yaml.load (if available) or ``None`` + + A callable that returns a schema dict from a file object. This is + used to load a schema from a data stream (YAML, et al.) + Typen =====
--- a/tests/test_schema.py Sat Jul 08 12:26:44 2023 +0200 +++ b/tests/test_schema.py Sat Jul 08 13:51:20 2023 +0200 @@ -140,7 +140,8 @@ obj = object() schema = object() settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) ctx = data_schema.Context( None, root_object=obj, root_schema=schema, settings=settings) self.assertEqual("<ROOT>", str(ctx)) @@ -152,7 +153,8 @@ obj = object() schema = object() settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) ctx = data_schema.Context( None, root_object=obj, root_schema=schema, settings=settings) self.assertTrue(ctx.is_root) @@ -167,7 +169,8 @@ def test_root_context_init_root_empty(self): settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) self.assertRaises( TypeError, data_schema.Context, None, key="key", settings=settings) @@ -177,7 +180,8 @@ def test_root_context_init_only_one_of_key_index(self): settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) root = data_schema.Context(None, settings=settings) self.assertRaises( ValueError, @@ -185,13 +189,15 @@ def test_root_context_init_exactly_one(self): settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) root = data_schema.Context(None, settings=settings) self.assertRaises(TypeError, data_schema.Context, root) def test_nonroot_rootobj_schema(self): settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) obj = object() schema = object() ctx = data_schema.Context( @@ -207,7 +213,8 @@ def test_str(self): settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) root = data_schema.Context(None, settings=settings) ctx1 = data_schema.Context(root, key="key1") ctx2 = data_schema.Context(ctx1, index=2) @@ -216,7 +223,8 @@ def test_repr(self): settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) root = data_schema.Context(None, settings=settings) ctx1 = data_schema.Context(root, key="key1") ctx2 = data_schema.Context(ctx1, index=2) @@ -225,7 +233,8 @@ def test_root(self): settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) root = data_schema.Context(None, settings=settings) self.assertTrue(root.is_root) self.assertTrue(root is root.root) @@ -244,9 +253,11 @@ def test_extra_settings_in_between(self): settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) settings2 = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) root = data_schema.Context(None, settings=settings) self.assertTrue(root.is_root) self.assertTrue(root is root.root) @@ -265,7 +276,8 @@ def test_key_xor_index(self): settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) root = data_schema.Context(None, settings=settings) self.assertRaises( ValueError, @@ -276,7 +288,8 @@ def test_keyindex_requires_key(self): settings = data_schema.ValidationSettings( - skip_keys=[], break_on_keynames_problems=True) + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) self.assertRaises( ValueError, data_schema.Context, @@ -1234,6 +1247,34 @@ schema, self._ctx) + def test_raise_no_schema_loader_available(self): + settings = data_schema.ValidationSettings( + skip_keys=[], break_on_keynames_problems=True, + schema_loader=None) + schema = data_schema._Schema( + None, True, {"$ref": "schema:file:/tmp/xxx#/"}) + ctx = data_schema.Context( + None, root_schema=schema, settings=settings) + self.assertRaises( + data_schema.SchemaError, + data_schema.process_schema_references, + schema, + ctx) + + def test_raise_schema_loader_available_but_invalid_basedir(self): + settings = data_schema.ValidationSettings( + skip_keys=[], break_on_keynames_problems=True, + schema_loader=data_schema.default_schema_loader) + schema = data_schema._Schema( + None, True, {"$ref": "schema:file:/tmp/xxx#/"}) + ctx = data_schema.Context( + None, root_schema=schema, settings=settings) + self.assertRaises( + TypeError, # no basedir given + data_schema.process_schema_references, + schema, + ctx) + def test_raise_if_scheme_ref_is_not_the_single_key(self): schema = data_schema._Schema( None, True, {"$ref": "schema:#/",