Source code for gemseo.third_party.fastjsonschema.ref_resolver

# -*- coding: utf-8 -*-
JSON Schema URI resolution scopes and dereferencing

Code adapted from

from __future__ import absolute_import, unicode_literals, with_statement

import contextlib
import json
import re

import requests

from gemseo.utils.py23_compat import PY3

from .exceptions import JsonSchemaException

if PY3:
    import urllib.parse as urlparse
    from urllib.request import urlopen
    from urllib.parse import unquote
    from urllib2 import urlopen
    import urlparse
    from urllib import unquote

[docs]def resolve_path(schema, fragment): """ Return definition from path. Path is unescaped according """ fragment = fragment.lstrip("/") parts = unquote(fragment).split("/") if fragment else [] for part in parts: part = part.replace("~1", "/").replace("~0", "~") if isinstance(schema, list): schema = schema[int(part)] elif part in schema: schema = schema[part] else: raise JsonSchemaException("Unresolvable ref: {}".format(part)) return schema
[docs]def normalize(uri): return urlparse.urlsplit(uri).geturl()
[docs]def resolve_remote(uri, handlers): """ Resolve a remote ``uri``. .. note:: Requests_ library is used to fetch ``http`` or ``https`` requests from the remote ``uri``, if handlers does not define otherwise. For unknown schemes urlib is used with UTF-8 encoding. .. _Requests: """ scheme = urlparse.urlsplit(uri).scheme if scheme in handlers: result = handlers[scheme](uri) elif scheme in ["http", "https"]: result = requests.get(uri).json() else: result = json.loads(urlopen(uri).read().decode("utf-8")) return result
[docs]class RefResolver(object): """ Resolve JSON References. """ # pylint: disable=dangerous-default-value,too-many-arguments def __init__(self, base_uri, schema, store={}, cache=True, handlers={}): """ `base_uri` is URI of the referring document from the `schema`. """ self.base_uri = base_uri self.resolution_scope = base_uri self.schema = schema = store self.cache = cache self.handlers = handlers self.walk(schema)
[docs] @classmethod def from_schema(cls, schema, handlers={}, **kwargs): """ Construct a resolver from a JSON schema object. """ return cls( schema.get("$id", schema.get("id", "")) if isinstance(schema, dict) else "", schema, handlers=handlers, **kwargs )
[docs] @contextlib.contextmanager def in_scope(self, scope): """ Context manager to handle current scope. """ old_scope = self.resolution_scope self.resolution_scope = urlparse.urljoin(old_scope, scope) try: yield finally: self.resolution_scope = old_scope
[docs] @contextlib.contextmanager def resolving(self, ref): """ Context manager which resolves a JSON ``ref`` and enters the resolution scope of this ref. """ new_uri = urlparse.urljoin(self.resolution_scope, ref) uri, fragment = urlparse.urldefrag(new_uri) if normalize(uri) in schema =[normalize(uri)] elif not uri or uri == self.base_uri: schema = self.schema else: schema = resolve_remote(uri, self.handlers) if self.cache:[normalize(uri)] = schema old_base_uri, old_schema = self.base_uri, self.schema self.base_uri, self.schema = uri, schema try: with self.in_scope(uri): yield resolve_path(schema, fragment) finally: self.base_uri, self.schema = old_base_uri, old_schema
[docs] def get_uri(self): return normalize(self.resolution_scope)
[docs] def get_scope_name(self): """ Get current scope and return it as a valid function name. """ name = "validate_" + unquote(self.resolution_scope).replace("~1", "_").replace( "~0", "_" ) name = re.sub(r"[:/#\.\-\%]", "_", name) name = name.lower().rstrip("_") return name
[docs] def walk(self, node): """ Walk thru schema and dereferencing ``id`` and ``$ref`` instances """ if isinstance(node, bool): pass elif "$ref" in node and isinstance(node["$ref"], unicode): ref = node["$ref"] node["$ref"] = urlparse.urljoin(self.resolution_scope, ref) elif "id" in node and isinstance(node["id"], unicode): with self.in_scope(node["id"]):[normalize(self.resolution_scope)] = node for _, item in node.items(): if isinstance(item, dict): self.walk(item) else: for _, item in node.items(): if isinstance(item, dict): self.walk(item)