# # This file was autogenerated using schema-salad-tool --codegen=python # from __future__ import absolute_import import os import re import traceback from typing import (Any, AnyStr, Callable, Dict, List, MutableMapping, MutableSequence, Union) import ruamel.yaml import six from ruamel.yaml.comments import CommentedBase, CommentedMap, CommentedSeq from typing_extensions import Text # pylint: disable=unused-import # move to a regular typing import when Python 3.3-3.6 is no longer supported lineno_re = re.compile(u"^(.*?:[0-9]+:[0-9]+: )(( *)(.*))") def _add_lc_filename(r, source): # type: (ruamel.yaml.comments.CommentedBase, AnyStr) -> None if isinstance(r, ruamel.yaml.comments.CommentedBase): r.lc.filename = source if isinstance(r, MutableSequence): for d in r: _add_lc_filename(d, source) elif isinstance(r, MutableMapping): for d in six.itervalues(r): _add_lc_filename(d, source) def relname(source): # type: (Text) -> Text if source.startswith("file://"): source = source[7:] source = os.path.relpath(source) return source def add_lc_filename(r, source): # type: (ruamel.yaml.comments.CommentedBase, Text) -> None _add_lc_filename(r, relname(source)) def reflow(text, maxline, shift=""): # type: (Text, int, Text) -> Text if maxline < 20: maxline = 20 if len(text) > maxline: sp = text.rfind(' ', 0, maxline) if sp < 1: sp = text.find(' ', sp+1) if sp == -1: sp = len(text) if sp < len(text): return "%s\n%s%s" % (text[0:sp], shift, reflow(text[sp+1:], maxline, shift)) return text def indent(v, nolead=False, shift=u" ", bullet=u" "): # type: (Text, bool, Text, Text) -> Text if nolead: return v.splitlines()[0] + u"\n".join([shift + l for l in v.splitlines()[1:]]) else: def lineno(i, l): # type: (int, Text) -> Text r = lineno_re.match(l) if r is not None: return r.group(1) + (bullet if i == 0 else shift) + r.group(2) else: return (bullet if i == 0 else shift) + l return u"\n".join([lineno(i, l) for i, l in enumerate(v.splitlines())]) def bullets(textlist, bul): # type: (List[Text], Text) -> Text if len(textlist) == 1: return textlist[0] else: return "\n".join(indent(t, bullet=bul) for t in textlist) def strip_dup_lineno(text, maxline=None): # type: (Text, int) -> Text if maxline is None: maxline = int(os.environ.get("COLUMNS", "100")) pre = None msg = [] maxno = 0 for l in text.splitlines(): g = lineno_re.match(l) if not g: continue maxno = max(maxno, len(g.group(1))) for l in text.splitlines(): g = lineno_re.match(l) if not g: msg.append(l) continue if g.group(1) != pre: shift = maxno + len(g.group(3)) g2 = reflow(g.group(2), maxline-shift, " " * shift) pre = g.group(1) msg.append(pre + " " * (maxno-len(g.group(1))) + g2) else: g2 = reflow(g.group(2), maxline-maxno, " " * (maxno+len(g.group(3)))) msg.append(" " * maxno + g2) return "\n".join(msg) def cmap(d, lc=None, fn=None): # type: (Union[int, float, str, Text, Dict, List], List[int], Text) -> Union[int, float, str, Text, CommentedMap, CommentedSeq] if lc is None: lc = [0, 0, 0, 0] if fn is None: fn = "test" if isinstance(d, CommentedMap): fn = d.lc.filename if hasattr(d.lc, "filename") else fn for k,v in six.iteritems(d): if k in d.lc.data: d[k] = cmap(v, lc=d.lc.data[k], fn=fn) else: d[k] = cmap(v, lc, fn=fn) return d if isinstance(d, CommentedSeq): fn = d.lc.filename if hasattr(d.lc, "filename") else fn for k,v in enumerate(d): if k in d.lc.data: d[k] = cmap(v, lc=d.lc.data[k], fn=fn) else: d[k] = cmap(v, lc, fn=fn) return d if isinstance(d, MutableMapping): cm = CommentedMap() for k in sorted(d.keys()): v = d[k] if isinstance(v, CommentedBase): uselc = [v.lc.line, v.lc.col, v.lc.line, v.lc.col] vfn = v.lc.filename if hasattr(v.lc, "filename") else fn else: uselc = lc vfn = fn cm[k] = cmap(v, lc=uselc, fn=vfn) cm.lc.add_kv_line_col(k, uselc) cm.lc.filename = fn return cm if isinstance(d, MutableSequence): cs = CommentedSeq() for k,v in enumerate(d): if isinstance(v, CommentedBase): uselc = [v.lc.line, v.lc.col, v.lc.line, v.lc.col] vfn = v.lc.filename if hasattr(v.lc, "filename") else fn else: uselc = lc vfn = fn cs.append(cmap(v, lc=uselc, fn=vfn)) cs.lc.add_kv_line_col(k, uselc) cs.lc.filename = fn return cs else: return d class SourceLine(object): def __init__(self, item, key=None, raise_type=six.text_type, include_traceback=False): # type: (Any, Any, Callable, bool) -> None self.item = item self.key = key self.raise_type = raise_type self.include_traceback = include_traceback def __enter__(self): # type: () -> SourceLine return self def __exit__(self, exc_type, # type: Any exc_value, # type: Any tb # type: Any ): # -> Any if not exc_value: return if self.include_traceback: raise self.makeError("\n".join(traceback.format_exception(exc_type, exc_value, tb))) else: raise self.makeError(six.text_type(exc_value)) def makeLead(self): # type: () -> Text if self.key is None or self.item.lc.data is None or self.key not in self.item.lc.data: return "%s:%i:%i:" % (self.item.lc.filename if hasattr(self.item.lc, "filename") else "", (self.item.lc.line or 0)+1, (self.item.lc.col or 0)+1) else: return "%s:%i:%i:" % (self.item.lc.filename if hasattr(self.item.lc, "filename") else "", (self.item.lc.data[self.key][0] or 0)+1, (self.item.lc.data[self.key][1] or 0)+1) def makeError(self, msg): # type: (Text) -> Any if not isinstance(self.item, ruamel.yaml.comments.CommentedBase): return self.raise_type(msg) errs = [] lead = self.makeLead() for m in msg.splitlines(): if bool(lineno_re.match(m)): errs.append(m) else: errs.append("%s %s" % (lead, m)) return self.raise_type("\n".join(errs)) import copy import re import uuid # pylint: disable=unused-import from typing import (Any, Dict, List, MutableMapping, MutableSequence, Sequence, Union) from ruamel import yaml from six import iteritems, string_types, text_type from six.moves import StringIO, urllib from typing_extensions import Text # pylint: disable=unused-import # move to a regular typing import when Python 3.3-3.6 is no longer supported class ValidationException(Exception): pass class Savable(object): pass class LoadingOptions(object): def __init__(self, fetcher=None, namespaces=None, fileuri=None, copyfrom=None, schemas=None): if copyfrom is not None: self.idx = copyfrom.idx if fetcher is None: fetcher = copyfrom.fetcher if fileuri is None: fileuri = copyfrom.fileuri if namespaces is None: namespaces = copyfrom.namespaces if namespaces is None: schemas = copyfrom.schemas else: self.idx = {} if fetcher is None: import os import requests from cachecontrol.wrapper import CacheControl from cachecontrol.caches import FileCache from schema_salad.ref_resolver import DefaultFetcher if "HOME" in os.environ: session = CacheControl( requests.Session(), cache=FileCache(os.path.join(os.environ["HOME"], ".cache", "salad"))) elif "TMPDIR" in os.environ: session = CacheControl( requests.Session(), cache=FileCache(os.path.join(os.environ["TMPDIR"], ".cache", "salad"))) else: session = CacheControl( requests.Session(), cache=FileCache("/tmp", ".cache", "salad")) self.fetcher = DefaultFetcher({}, session) else: self.fetcher = fetcher self.fileuri = fileuri self.vocab = _vocab self.rvocab = _rvocab self.namespaces = namespaces self.schemas = schemas if namespaces is not None: self.vocab = self.vocab.copy() self.rvocab = self.rvocab.copy() for k,v in iteritems(namespaces): self.vocab[k] = v self.rvocab[v] = k def load_field(val, fieldtype, baseuri, loadingOptions): if isinstance(val, MutableMapping): if "$import" in val: return _document_load_by_url(fieldtype, loadingOptions.fetcher.urljoin(loadingOptions.fileuri, val["$import"]), loadingOptions) elif "$include" in val: val = loadingOptions.fetcher.fetch_text(loadingOptions.fetcher.urljoin(loadingOptions.fileuri, val["$include"])) return fieldtype.load(val, baseuri, loadingOptions) def save(val, top=True, base_url="", relative_uris=True): if isinstance(val, Savable): return val.save(top=top, base_url=base_url, relative_uris=relative_uris) if isinstance(val, MutableSequence): return [save(v, top=False, base_url=base_url, relative_uris=relative_uris) for v in val] return val def expand_url(url, # type: Union[str, Text] base_url, # type: Union[str, Text] loadingOptions, # type: LoadingOptions scoped_id=False, # type: bool vocab_term=False, # type: bool scoped_ref=None # type: int ): # type: (...) -> Text if not isinstance(url, string_types): return url url = Text(url) if url in (u"@id", u"@type"): return url if vocab_term and url in loadingOptions.vocab: return url if bool(loadingOptions.vocab) and u":" in url: prefix = url.split(u":")[0] if prefix in loadingOptions.vocab: url = loadingOptions.vocab[prefix] + url[len(prefix) + 1:] split = urllib.parse.urlsplit(url) if ((bool(split.scheme) and split.scheme in [u'http', u'https', u'file']) or url.startswith(u"$(") or url.startswith(u"${")): pass elif scoped_id and not bool(split.fragment): splitbase = urllib.parse.urlsplit(base_url) frg = u"" if bool(splitbase.fragment): frg = splitbase.fragment + u"/" + split.path else: frg = split.path pt = splitbase.path if splitbase.path != '' else "/" url = urllib.parse.urlunsplit( (splitbase.scheme, splitbase.netloc, pt, splitbase.query, frg)) elif scoped_ref is not None and not bool(split.fragment): splitbase = urllib.parse.urlsplit(base_url) sp = splitbase.fragment.split(u"/") n = scoped_ref while n > 0 and len(sp) > 0: sp.pop() n -= 1 sp.append(url) url = urllib.parse.urlunsplit(( splitbase.scheme, splitbase.netloc, splitbase.path, splitbase.query, u"/".join(sp))) else: url = loadingOptions.fetcher.urljoin(base_url, url) if vocab_term: split = urllib.parse.urlsplit(url) if bool(split.scheme): if url in loadingOptions.rvocab: return loadingOptions.rvocab[url] else: raise ValidationException("Term '%s' not in vocabulary" % url) return url class _Loader(object): def load(self, doc, baseuri, loadingOptions, docRoot=None): # type: (Any, Text, LoadingOptions, Union[Text, None]) -> Any pass class _AnyLoader(_Loader): def load(self, doc, baseuri, loadingOptions, docRoot=None): if doc is not None: return doc raise ValidationException("Expected non-null") class _PrimitiveLoader(_Loader): def __init__(self, tp): # type: (Union[type, Sequence[type]]) -> None self.tp = tp def load(self, doc, baseuri, loadingOptions, docRoot=None): if not isinstance(doc, self.tp): raise ValidationException("Expected a %s but got %s" % (self.tp, type(doc))) return doc def __repr__(self): return str(self.tp) class _ArrayLoader(_Loader): def __init__(self, items): # type: (_Loader) -> None self.items = items def load(self, doc, baseuri, loadingOptions, docRoot=None): if not isinstance(doc, MutableSequence): raise ValidationException("Expected a list") r = [] errors = [] for i in range(0, len(doc)): try: lf = load_field(doc[i], _UnionLoader((self, self.items)), baseuri, loadingOptions) if isinstance(lf, MutableSequence): r.extend(lf) else: r.append(lf) except ValidationException as e: errors.append(SourceLine(doc, i, str).makeError(text_type(e))) if errors: raise ValidationException("\n".join(errors)) return r def __repr__(self): return "array<%s>" % self.items class _EnumLoader(_Loader): def __init__(self, symbols): # type: (Sequence[Text]) -> None self.symbols = symbols def load(self, doc, baseuri, loadingOptions, docRoot=None): if doc in self.symbols: return doc else: raise ValidationException("Expected one of %s" % (self.symbols,)) class _RecordLoader(_Loader): def __init__(self, classtype): # type: (type) -> None self.classtype = classtype def load(self, doc, baseuri, loadingOptions, docRoot=None): if not isinstance(doc, MutableMapping): raise ValidationException("Expected a dict") return self.classtype(doc, baseuri, loadingOptions, docRoot=docRoot) def __repr__(self): return str(self.classtype) class _UnionLoader(_Loader): def __init__(self, alternates): # type: (Sequence[_Loader]) -> None self.alternates = alternates def load(self, doc, baseuri, loadingOptions, docRoot=None): errors = [] for t in self.alternates: try: return t.load(doc, baseuri, loadingOptions, docRoot=docRoot) except ValidationException as e: errors.append("tried %s but\n%s" % (t, indent(str(e)))) raise ValidationException(bullets(errors, "- ")) def __repr__(self): return " | ".join(str(a) for a in self.alternates) class _URILoader(_Loader): def __init__(self, inner, scoped_id, vocab_term, scoped_ref): # type: (_Loader, bool, bool, Union[int, None]) -> None self.inner = inner self.scoped_id = scoped_id self.vocab_term = vocab_term self.scoped_ref = scoped_ref def load(self, doc, baseuri, loadingOptions, docRoot=None): if isinstance(doc, MutableSequence): doc = [expand_url(i, baseuri, loadingOptions, self.scoped_id, self.vocab_term, self.scoped_ref) for i in doc] if isinstance(doc, string_types): doc = expand_url(doc, baseuri, loadingOptions, self.scoped_id, self.vocab_term, self.scoped_ref) return self.inner.load(doc, baseuri, loadingOptions) class _TypeDSLLoader(_Loader): typeDSLregex = re.compile(r"^([^[?]+)(\[\])?(\?)?$") def __init__(self, inner, refScope): # type: (_Loader, Union[int, None]) -> None self.inner = inner self.refScope = refScope def resolve(self, doc, baseuri, loadingOptions): m = self.typeDSLregex.match(doc) if m: first = expand_url(m.group(1), baseuri, loadingOptions, False, True, self.refScope) second = third = None if bool(m.group(2)): second = {"type": "array", "items": first} #second = CommentedMap((("type", "array"), # ("items", first))) #second.lc.add_kv_line_col("type", lc) #second.lc.add_kv_line_col("items", lc) #second.lc.filename = filename if bool(m.group(3)): third = [u"null", second or first] #third = CommentedSeq([u"null", second or first]) #third.lc.add_kv_line_col(0, lc) #third.lc.add_kv_line_col(1, lc) #third.lc.filename = filename doc = third or second or first return doc def load(self, doc, baseuri, loadingOptions, docRoot=None): if isinstance(doc, MutableSequence): r = [] for d in doc: if isinstance(d, string_types): resolved = self.resolve(d, baseuri, loadingOptions) if isinstance(resolved, MutableSequence): for i in resolved: if i not in r: r.append(i) else: if resolved not in r: r.append(resolved) else: r.append(d) doc = r elif isinstance(doc, string_types): doc = self.resolve(doc, baseuri, loadingOptions) return self.inner.load(doc, baseuri, loadingOptions) class _IdMapLoader(_Loader): def __init__(self, inner, mapSubject, mapPredicate): # type: (_Loader, Text, Union[Text, None]) -> None self.inner = inner self.mapSubject = mapSubject self.mapPredicate = mapPredicate def load(self, doc, baseuri, loadingOptions, docRoot=None): if isinstance(doc, MutableMapping): r = [] for k in sorted(doc.keys()): val = doc[k] if isinstance(val, MutableMapping): v = copy.copy(val) if hasattr(val, 'lc'): v.lc.data = val.lc.data v.lc.filename = val.lc.filename else: if self.mapPredicate: v = {self.mapPredicate: val} else: raise ValidationException("No mapPredicate") v[self.mapSubject] = k r.append(v) doc = r return self.inner.load(doc, baseuri, loadingOptions) def _document_load(loader, doc, baseuri, loadingOptions): if isinstance(doc, string_types): return _document_load_by_url(loader, loadingOptions.fetcher.urljoin(baseuri, doc), loadingOptions) if isinstance(doc, MutableMapping): if "$namespaces" in doc: loadingOptions = LoadingOptions(copyfrom=loadingOptions, namespaces=doc["$namespaces"]) doc = {k: v for k,v in doc.items() if k != "$namespaces"} if "$schemas" in doc: loadingOptions = LoadingOptions(copyfrom=loadingOptions, schemas=doc["$schemas"]) doc = {k: v for k,v in doc.items() if k != "$schemas"} if "$base" in doc: baseuri = doc["$base"] if "$graph" in doc: return loader.load(doc["$graph"], baseuri, loadingOptions) else: return loader.load(doc, baseuri, loadingOptions, docRoot=baseuri) if isinstance(doc, MutableSequence): return loader.load(doc, baseuri, loadingOptions) raise ValidationException() def _document_load_by_url(loader, url, loadingOptions): if url in loadingOptions.idx: return _document_load(loader, loadingOptions.idx[url], url, loadingOptions) text = loadingOptions.fetcher.fetch_text(url) if isinstance(text, bytes): textIO = StringIO(text.decode('utf-8')) else: textIO = StringIO(text) textIO.name = url # type: ignore result = yaml.round_trip_load(textIO, preserve_quotes=True) add_lc_filename(result, url) loadingOptions.idx[url] = result loadingOptions = LoadingOptions(copyfrom=loadingOptions, fileuri=url) return _document_load(loader, result, url, loadingOptions) def file_uri(path, split_frag=False): # type: (str, bool) -> str if path.startswith("file://"): return path if split_frag: pathsp = path.split("#", 2) frag = "#" + urllib.parse.quote(str(pathsp[1])) if len(pathsp) == 2 else "" urlpath = urllib.request.pathname2url(str(pathsp[0])) else: urlpath = urllib.request.pathname2url(path) frag = "" if urlpath.startswith("//"): return "file:%s%s" % (urlpath, frag) else: return "file://%s%s" % (urlpath, frag) def prefix_url(url, namespaces): for k,v in namespaces.items(): if url.startswith(v): return k+":"+url[len(v):] return url def save_relative_uri(uri, base_url, scoped_id, ref_scope, relative_uris): if not relative_uris: return uri if isinstance(uri, MutableSequence): return [save_relative_uri(u, base_url, scoped_id, ref_scope, relative_uris) for u in uri] elif isinstance(uri, text_type): urisplit = urllib.parse.urlsplit(uri) basesplit = urllib.parse.urlsplit(base_url) if urisplit.scheme == basesplit.scheme and urisplit.netloc == basesplit.netloc: if urisplit.path != basesplit.path: p = os.path.relpath(urisplit.path, os.path.dirname(basesplit.path)) if urisplit.fragment: p = p + "#" + urisplit.fragment return p basefrag = basesplit.fragment+"/" if ref_scope: sp = basefrag.split("/") i = 0 while i < ref_scope: sp.pop() i += 1 basefrag = "/".join(sp) if urisplit.fragment.startswith(basefrag): return urisplit.fragment[len(basefrag):] else: return urisplit.fragment return uri else: return save(uri, top=False, base_url=base_url) class Documented(Savable): pass class RecordField(Documented): """ A field of a record. """ def __init__(self, _doc, baseuri, loadingOptions, docRoot=None): doc = copy.copy(_doc) if hasattr(_doc, 'lc'): doc.lc.data = _doc.lc.data doc.lc.filename = _doc.lc.filename errors = [] self.loadingOptions = loadingOptions if 'name' in doc: try: self.name = load_field(doc.get('name'), uri_strtype_True_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'name', str).makeError("the `name` field is not valid because:\n"+str(e))) else: self.name = None if self.name is None: if docRoot is not None: self.name = docRoot else: raise ValidationException("Missing name") baseuri = self.name if 'doc' in doc: try: self.doc = load_field(doc.get('doc'), union_of_None_type_or_strtype_or_array_of_strtype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'doc', str).makeError("the `doc` field is not valid because:\n"+str(e))) else: self.doc = None try: self.type = load_field(doc.get('type'), typedsl_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_or_array_of_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_2, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'type', str).makeError("the `type` field is not valid because:\n"+str(e))) self.extension_fields = {} for k in doc.keys(): if k not in self.attrs: if ":" in k: ex = expand_url(k, u"", loadingOptions, scoped_id=False, vocab_term=False) self.extension_fields[ex] = doc[k] else: errors.append(SourceLine(doc, k, str).makeError("invalid field `%s`, expected one of: `doc`, `name`, `type`" % (k))) break if errors: raise ValidationException("Trying 'RecordField'\n"+"\n".join(errors)) def save(self, top=False, base_url="", relative_uris=True): r = {} for ef in self.extension_fields: r[prefix_url(ef, self.loadingOptions.vocab)] = self.extension_fields[ef] if self.name is not None: u = save_relative_uri(self.name, base_url, True, None, relative_uris) if u: r['name'] = u if self.doc is not None: r['doc'] = save(self.doc, top=False, base_url=self.name, relative_uris=relative_uris) if self.type is not None: r['type'] = save(self.type, top=False, base_url=self.name, relative_uris=relative_uris) if top and self.loadingOptions.namespaces: r["$namespaces"] = self.loadingOptions.namespaces return r attrs = frozenset(['doc', 'name', 'type']) class RecordSchema(Savable): def __init__(self, _doc, baseuri, loadingOptions, docRoot=None): doc = copy.copy(_doc) if hasattr(_doc, 'lc'): doc.lc.data = _doc.lc.data doc.lc.filename = _doc.lc.filename errors = [] self.loadingOptions = loadingOptions if 'fields' in doc: try: self.fields = load_field(doc.get('fields'), idmap_fields_union_of_None_type_or_array_of_RecordFieldLoader, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'fields', str).makeError("the `fields` field is not valid because:\n"+str(e))) else: self.fields = None try: self.type = load_field(doc.get('type'), typedsl_enum_d9cba076fca539106791a4f46d198c7fcfbdb779Loader_2, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'type', str).makeError("the `type` field is not valid because:\n"+str(e))) self.extension_fields = {} for k in doc.keys(): if k not in self.attrs: if ":" in k: ex = expand_url(k, u"", loadingOptions, scoped_id=False, vocab_term=False) self.extension_fields[ex] = doc[k] else: errors.append(SourceLine(doc, k, str).makeError("invalid field `%s`, expected one of: `fields`, `type`" % (k))) break if errors: raise ValidationException("Trying 'RecordSchema'\n"+"\n".join(errors)) def save(self, top=False, base_url="", relative_uris=True): r = {} for ef in self.extension_fields: r[prefix_url(ef, self.loadingOptions.vocab)] = self.extension_fields[ef] if self.fields is not None: r['fields'] = save(self.fields, top=False, base_url=base_url, relative_uris=relative_uris) if self.type is not None: r['type'] = save(self.type, top=False, base_url=base_url, relative_uris=relative_uris) if top and self.loadingOptions.namespaces: r["$namespaces"] = self.loadingOptions.namespaces return r attrs = frozenset(['fields', 'type']) class EnumSchema(Savable): """ Define an enumerated type. """ def __init__(self, _doc, baseuri, loadingOptions, docRoot=None): doc = copy.copy(_doc) if hasattr(_doc, 'lc'): doc.lc.data = _doc.lc.data doc.lc.filename = _doc.lc.filename errors = [] self.loadingOptions = loadingOptions try: self.symbols = load_field(doc.get('symbols'), uri_array_of_strtype_True_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'symbols', str).makeError("the `symbols` field is not valid because:\n"+str(e))) try: self.type = load_field(doc.get('type'), typedsl_enum_d961d79c225752b9fadb617367615ab176b47d77Loader_2, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'type', str).makeError("the `type` field is not valid because:\n"+str(e))) self.extension_fields = {} for k in doc.keys(): if k not in self.attrs: if ":" in k: ex = expand_url(k, u"", loadingOptions, scoped_id=False, vocab_term=False) self.extension_fields[ex] = doc[k] else: errors.append(SourceLine(doc, k, str).makeError("invalid field `%s`, expected one of: `symbols`, `type`" % (k))) break if errors: raise ValidationException("Trying 'EnumSchema'\n"+"\n".join(errors)) def save(self, top=False, base_url="", relative_uris=True): r = {} for ef in self.extension_fields: r[prefix_url(ef, self.loadingOptions.vocab)] = self.extension_fields[ef] if self.symbols is not None: u = save_relative_uri(self.symbols, base_url, True, None, relative_uris) if u: r['symbols'] = u if self.type is not None: r['type'] = save(self.type, top=False, base_url=base_url, relative_uris=relative_uris) if top and self.loadingOptions.namespaces: r["$namespaces"] = self.loadingOptions.namespaces return r attrs = frozenset(['symbols', 'type']) class ArraySchema(Savable): def __init__(self, _doc, baseuri, loadingOptions, docRoot=None): doc = copy.copy(_doc) if hasattr(_doc, 'lc'): doc.lc.data = _doc.lc.data doc.lc.filename = _doc.lc.filename errors = [] self.loadingOptions = loadingOptions try: self.items = load_field(doc.get('items'), uri_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_or_array_of_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_False_True_2, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'items', str).makeError("the `items` field is not valid because:\n"+str(e))) try: self.type = load_field(doc.get('type'), typedsl_enum_d062602be0b4b8fd33e69e29a841317b6ab665bcLoader_2, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'type', str).makeError("the `type` field is not valid because:\n"+str(e))) self.extension_fields = {} for k in doc.keys(): if k not in self.attrs: if ":" in k: ex = expand_url(k, u"", loadingOptions, scoped_id=False, vocab_term=False) self.extension_fields[ex] = doc[k] else: errors.append(SourceLine(doc, k, str).makeError("invalid field `%s`, expected one of: `items`, `type`" % (k))) break if errors: raise ValidationException("Trying 'ArraySchema'\n"+"\n".join(errors)) def save(self, top=False, base_url="", relative_uris=True): r = {} for ef in self.extension_fields: r[prefix_url(ef, self.loadingOptions.vocab)] = self.extension_fields[ef] if self.items is not None: u = save_relative_uri(self.items, base_url, False, 2, relative_uris) if u: r['items'] = u if self.type is not None: r['type'] = save(self.type, top=False, base_url=base_url, relative_uris=relative_uris) if top and self.loadingOptions.namespaces: r["$namespaces"] = self.loadingOptions.namespaces return r attrs = frozenset(['items', 'type']) class JsonldPredicate(Savable): """ Attached to a record field to define how the parent record field is handled for URI resolution and JSON-LD context generation. """ def __init__(self, _doc, baseuri, loadingOptions, docRoot=None): doc = copy.copy(_doc) if hasattr(_doc, 'lc'): doc.lc.data = _doc.lc.data doc.lc.filename = _doc.lc.filename errors = [] self.loadingOptions = loadingOptions if '_id' in doc: try: self._id = load_field(doc.get('_id'), uri_union_of_None_type_or_strtype_True_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, '_id', str).makeError("the `_id` field is not valid because:\n"+str(e))) else: self._id = None if '_type' in doc: try: self._type = load_field(doc.get('_type'), union_of_None_type_or_strtype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, '_type', str).makeError("the `_type` field is not valid because:\n"+str(e))) else: self._type = None if '_container' in doc: try: self._container = load_field(doc.get('_container'), union_of_None_type_or_strtype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, '_container', str).makeError("the `_container` field is not valid because:\n"+str(e))) else: self._container = None if 'identity' in doc: try: self.identity = load_field(doc.get('identity'), union_of_None_type_or_booltype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'identity', str).makeError("the `identity` field is not valid because:\n"+str(e))) else: self.identity = None if 'noLinkCheck' in doc: try: self.noLinkCheck = load_field(doc.get('noLinkCheck'), union_of_None_type_or_booltype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'noLinkCheck', str).makeError("the `noLinkCheck` field is not valid because:\n"+str(e))) else: self.noLinkCheck = None if 'mapSubject' in doc: try: self.mapSubject = load_field(doc.get('mapSubject'), union_of_None_type_or_strtype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'mapSubject', str).makeError("the `mapSubject` field is not valid because:\n"+str(e))) else: self.mapSubject = None if 'mapPredicate' in doc: try: self.mapPredicate = load_field(doc.get('mapPredicate'), union_of_None_type_or_strtype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'mapPredicate', str).makeError("the `mapPredicate` field is not valid because:\n"+str(e))) else: self.mapPredicate = None if 'refScope' in doc: try: self.refScope = load_field(doc.get('refScope'), union_of_None_type_or_inttype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'refScope', str).makeError("the `refScope` field is not valid because:\n"+str(e))) else: self.refScope = None if 'typeDSL' in doc: try: self.typeDSL = load_field(doc.get('typeDSL'), union_of_None_type_or_booltype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'typeDSL', str).makeError("the `typeDSL` field is not valid because:\n"+str(e))) else: self.typeDSL = None if 'subscope' in doc: try: self.subscope = load_field(doc.get('subscope'), union_of_None_type_or_strtype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'subscope', str).makeError("the `subscope` field is not valid because:\n"+str(e))) else: self.subscope = None self.extension_fields = {} for k in doc.keys(): if k not in self.attrs: if ":" in k: ex = expand_url(k, u"", loadingOptions, scoped_id=False, vocab_term=False) self.extension_fields[ex] = doc[k] else: errors.append(SourceLine(doc, k, str).makeError("invalid field `%s`, expected one of: `_id`, `_type`, `_container`, `identity`, `noLinkCheck`, `mapSubject`, `mapPredicate`, `refScope`, `typeDSL`, `subscope`" % (k))) break if errors: raise ValidationException("Trying 'JsonldPredicate'\n"+"\n".join(errors)) def save(self, top=False, base_url="", relative_uris=True): r = {} for ef in self.extension_fields: r[prefix_url(ef, self.loadingOptions.vocab)] = self.extension_fields[ef] if self._id is not None: u = save_relative_uri(self._id, base_url, True, None, relative_uris) if u: r['_id'] = u if self._type is not None: r['_type'] = save(self._type, top=False, base_url=base_url, relative_uris=relative_uris) if self._container is not None: r['_container'] = save(self._container, top=False, base_url=base_url, relative_uris=relative_uris) if self.identity is not None: r['identity'] = save(self.identity, top=False, base_url=base_url, relative_uris=relative_uris) if self.noLinkCheck is not None: r['noLinkCheck'] = save(self.noLinkCheck, top=False, base_url=base_url, relative_uris=relative_uris) if self.mapSubject is not None: r['mapSubject'] = save(self.mapSubject, top=False, base_url=base_url, relative_uris=relative_uris) if self.mapPredicate is not None: r['mapPredicate'] = save(self.mapPredicate, top=False, base_url=base_url, relative_uris=relative_uris) if self.refScope is not None: r['refScope'] = save(self.refScope, top=False, base_url=base_url, relative_uris=relative_uris) if self.typeDSL is not None: r['typeDSL'] = save(self.typeDSL, top=False, base_url=base_url, relative_uris=relative_uris) if self.subscope is not None: r['subscope'] = save(self.subscope, top=False, base_url=base_url, relative_uris=relative_uris) if top and self.loadingOptions.namespaces: r["$namespaces"] = self.loadingOptions.namespaces return r attrs = frozenset(['_id', '_type', '_container', 'identity', 'noLinkCheck', 'mapSubject', 'mapPredicate', 'refScope', 'typeDSL', 'subscope']) class SpecializeDef(Savable): def __init__(self, _doc, baseuri, loadingOptions, docRoot=None): doc = copy.copy(_doc) if hasattr(_doc, 'lc'): doc.lc.data = _doc.lc.data doc.lc.filename = _doc.lc.filename errors = [] self.loadingOptions = loadingOptions try: self.specializeFrom = load_field(doc.get('specializeFrom'), uri_strtype_False_False_1, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'specializeFrom', str).makeError("the `specializeFrom` field is not valid because:\n"+str(e))) try: self.specializeTo = load_field(doc.get('specializeTo'), uri_strtype_False_False_1, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'specializeTo', str).makeError("the `specializeTo` field is not valid because:\n"+str(e))) self.extension_fields = {} for k in doc.keys(): if k not in self.attrs: if ":" in k: ex = expand_url(k, u"", loadingOptions, scoped_id=False, vocab_term=False) self.extension_fields[ex] = doc[k] else: errors.append(SourceLine(doc, k, str).makeError("invalid field `%s`, expected one of: `specializeFrom`, `specializeTo`" % (k))) break if errors: raise ValidationException("Trying 'SpecializeDef'\n"+"\n".join(errors)) def save(self, top=False, base_url="", relative_uris=True): r = {} for ef in self.extension_fields: r[prefix_url(ef, self.loadingOptions.vocab)] = self.extension_fields[ef] if self.specializeFrom is not None: u = save_relative_uri(self.specializeFrom, base_url, False, 1, relative_uris) if u: r['specializeFrom'] = u if self.specializeTo is not None: u = save_relative_uri(self.specializeTo, base_url, False, 1, relative_uris) if u: r['specializeTo'] = u if top and self.loadingOptions.namespaces: r["$namespaces"] = self.loadingOptions.namespaces return r attrs = frozenset(['specializeFrom', 'specializeTo']) class NamedType(Savable): pass class DocType(Documented): pass class SchemaDefinedType(DocType): """ Abstract base for schema-defined types. """ pass class SaladRecordField(RecordField): """ A field of a record. """ def __init__(self, _doc, baseuri, loadingOptions, docRoot=None): doc = copy.copy(_doc) if hasattr(_doc, 'lc'): doc.lc.data = _doc.lc.data doc.lc.filename = _doc.lc.filename errors = [] self.loadingOptions = loadingOptions if 'name' in doc: try: self.name = load_field(doc.get('name'), uri_strtype_True_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'name', str).makeError("the `name` field is not valid because:\n"+str(e))) else: self.name = None if self.name is None: if docRoot is not None: self.name = docRoot else: raise ValidationException("Missing name") baseuri = self.name if 'doc' in doc: try: self.doc = load_field(doc.get('doc'), union_of_None_type_or_strtype_or_array_of_strtype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'doc', str).makeError("the `doc` field is not valid because:\n"+str(e))) else: self.doc = None try: self.type = load_field(doc.get('type'), typedsl_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_or_array_of_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_2, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'type', str).makeError("the `type` field is not valid because:\n"+str(e))) if 'jsonldPredicate' in doc: try: self.jsonldPredicate = load_field(doc.get('jsonldPredicate'), union_of_None_type_or_strtype_or_JsonldPredicateLoader, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'jsonldPredicate', str).makeError("the `jsonldPredicate` field is not valid because:\n"+str(e))) else: self.jsonldPredicate = None if 'default' in doc: try: self.default = load_field(doc.get('default'), union_of_None_type_or_Any_type, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'default', str).makeError("the `default` field is not valid because:\n"+str(e))) else: self.default = None self.extension_fields = {} for k in doc.keys(): if k not in self.attrs: if ":" in k: ex = expand_url(k, u"", loadingOptions, scoped_id=False, vocab_term=False) self.extension_fields[ex] = doc[k] else: errors.append(SourceLine(doc, k, str).makeError("invalid field `%s`, expected one of: `doc`, `name`, `type`, `jsonldPredicate`, `default`" % (k))) break if errors: raise ValidationException("Trying 'SaladRecordField'\n"+"\n".join(errors)) def save(self, top=False, base_url="", relative_uris=True): r = {} for ef in self.extension_fields: r[prefix_url(ef, self.loadingOptions.vocab)] = self.extension_fields[ef] if self.name is not None: u = save_relative_uri(self.name, base_url, True, None, relative_uris) if u: r['name'] = u if self.doc is not None: r['doc'] = save(self.doc, top=False, base_url=self.name, relative_uris=relative_uris) if self.type is not None: r['type'] = save(self.type, top=False, base_url=self.name, relative_uris=relative_uris) if self.jsonldPredicate is not None: r['jsonldPredicate'] = save(self.jsonldPredicate, top=False, base_url=self.name, relative_uris=relative_uris) if self.default is not None: r['default'] = save(self.default, top=False, base_url=self.name, relative_uris=relative_uris) if top and self.loadingOptions.namespaces: r["$namespaces"] = self.loadingOptions.namespaces return r attrs = frozenset(['doc', 'name', 'type', 'jsonldPredicate', 'default']) class SaladRecordSchema(NamedType, RecordSchema, SchemaDefinedType): def __init__(self, _doc, baseuri, loadingOptions, docRoot=None): doc = copy.copy(_doc) if hasattr(_doc, 'lc'): doc.lc.data = _doc.lc.data doc.lc.filename = _doc.lc.filename errors = [] self.loadingOptions = loadingOptions if 'name' in doc: try: self.name = load_field(doc.get('name'), uri_strtype_True_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'name', str).makeError("the `name` field is not valid because:\n"+str(e))) else: self.name = None if self.name is None: if docRoot is not None: self.name = docRoot else: raise ValidationException("Missing name") baseuri = self.name if 'inVocab' in doc: try: self.inVocab = load_field(doc.get('inVocab'), union_of_None_type_or_booltype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'inVocab', str).makeError("the `inVocab` field is not valid because:\n"+str(e))) else: self.inVocab = None if 'fields' in doc: try: self.fields = load_field(doc.get('fields'), idmap_fields_union_of_None_type_or_array_of_SaladRecordFieldLoader, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'fields', str).makeError("the `fields` field is not valid because:\n"+str(e))) else: self.fields = None try: self.type = load_field(doc.get('type'), typedsl_enum_d9cba076fca539106791a4f46d198c7fcfbdb779Loader_2, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'type', str).makeError("the `type` field is not valid because:\n"+str(e))) if 'doc' in doc: try: self.doc = load_field(doc.get('doc'), union_of_None_type_or_strtype_or_array_of_strtype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'doc', str).makeError("the `doc` field is not valid because:\n"+str(e))) else: self.doc = None if 'docParent' in doc: try: self.docParent = load_field(doc.get('docParent'), uri_union_of_None_type_or_strtype_False_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'docParent', str).makeError("the `docParent` field is not valid because:\n"+str(e))) else: self.docParent = None if 'docChild' in doc: try: self.docChild = load_field(doc.get('docChild'), uri_union_of_None_type_or_strtype_or_array_of_strtype_False_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'docChild', str).makeError("the `docChild` field is not valid because:\n"+str(e))) else: self.docChild = None if 'docAfter' in doc: try: self.docAfter = load_field(doc.get('docAfter'), uri_union_of_None_type_or_strtype_False_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'docAfter', str).makeError("the `docAfter` field is not valid because:\n"+str(e))) else: self.docAfter = None if 'jsonldPredicate' in doc: try: self.jsonldPredicate = load_field(doc.get('jsonldPredicate'), union_of_None_type_or_strtype_or_JsonldPredicateLoader, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'jsonldPredicate', str).makeError("the `jsonldPredicate` field is not valid because:\n"+str(e))) else: self.jsonldPredicate = None if 'documentRoot' in doc: try: self.documentRoot = load_field(doc.get('documentRoot'), union_of_None_type_or_booltype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'documentRoot', str).makeError("the `documentRoot` field is not valid because:\n"+str(e))) else: self.documentRoot = None if 'abstract' in doc: try: self.abstract = load_field(doc.get('abstract'), union_of_None_type_or_booltype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'abstract', str).makeError("the `abstract` field is not valid because:\n"+str(e))) else: self.abstract = None if 'extends' in doc: try: self.extends = load_field(doc.get('extends'), uri_union_of_None_type_or_strtype_or_array_of_strtype_False_False_1, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'extends', str).makeError("the `extends` field is not valid because:\n"+str(e))) else: self.extends = None if 'specialize' in doc: try: self.specialize = load_field(doc.get('specialize'), idmap_specialize_union_of_None_type_or_array_of_SpecializeDefLoader, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'specialize', str).makeError("the `specialize` field is not valid because:\n"+str(e))) else: self.specialize = None self.extension_fields = {} for k in doc.keys(): if k not in self.attrs: if ":" in k: ex = expand_url(k, u"", loadingOptions, scoped_id=False, vocab_term=False) self.extension_fields[ex] = doc[k] else: errors.append(SourceLine(doc, k, str).makeError("invalid field `%s`, expected one of: `name`, `inVocab`, `fields`, `type`, `doc`, `docParent`, `docChild`, `docAfter`, `jsonldPredicate`, `documentRoot`, `abstract`, `extends`, `specialize`" % (k))) break if errors: raise ValidationException("Trying 'SaladRecordSchema'\n"+"\n".join(errors)) def save(self, top=False, base_url="", relative_uris=True): r = {} for ef in self.extension_fields: r[prefix_url(ef, self.loadingOptions.vocab)] = self.extension_fields[ef] if self.name is not None: u = save_relative_uri(self.name, base_url, True, None, relative_uris) if u: r['name'] = u if self.inVocab is not None: r['inVocab'] = save(self.inVocab, top=False, base_url=self.name, relative_uris=relative_uris) if self.fields is not None: r['fields'] = save(self.fields, top=False, base_url=self.name, relative_uris=relative_uris) if self.type is not None: r['type'] = save(self.type, top=False, base_url=self.name, relative_uris=relative_uris) if self.doc is not None: r['doc'] = save(self.doc, top=False, base_url=self.name, relative_uris=relative_uris) if self.docParent is not None: u = save_relative_uri(self.docParent, self.name, False, None, relative_uris) if u: r['docParent'] = u if self.docChild is not None: u = save_relative_uri(self.docChild, self.name, False, None, relative_uris) if u: r['docChild'] = u if self.docAfter is not None: u = save_relative_uri(self.docAfter, self.name, False, None, relative_uris) if u: r['docAfter'] = u if self.jsonldPredicate is not None: r['jsonldPredicate'] = save(self.jsonldPredicate, top=False, base_url=self.name, relative_uris=relative_uris) if self.documentRoot is not None: r['documentRoot'] = save(self.documentRoot, top=False, base_url=self.name, relative_uris=relative_uris) if self.abstract is not None: r['abstract'] = save(self.abstract, top=False, base_url=self.name, relative_uris=relative_uris) if self.extends is not None: u = save_relative_uri(self.extends, self.name, False, 1, relative_uris) if u: r['extends'] = u if self.specialize is not None: r['specialize'] = save(self.specialize, top=False, base_url=self.name, relative_uris=relative_uris) if top and self.loadingOptions.namespaces: r["$namespaces"] = self.loadingOptions.namespaces return r attrs = frozenset(['name', 'inVocab', 'fields', 'type', 'doc', 'docParent', 'docChild', 'docAfter', 'jsonldPredicate', 'documentRoot', 'abstract', 'extends', 'specialize']) class SaladEnumSchema(NamedType, EnumSchema, SchemaDefinedType): """ Define an enumerated type. """ def __init__(self, _doc, baseuri, loadingOptions, docRoot=None): doc = copy.copy(_doc) if hasattr(_doc, 'lc'): doc.lc.data = _doc.lc.data doc.lc.filename = _doc.lc.filename errors = [] self.loadingOptions = loadingOptions if 'name' in doc: try: self.name = load_field(doc.get('name'), uri_strtype_True_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'name', str).makeError("the `name` field is not valid because:\n"+str(e))) else: self.name = None if self.name is None: if docRoot is not None: self.name = docRoot else: raise ValidationException("Missing name") baseuri = self.name if 'inVocab' in doc: try: self.inVocab = load_field(doc.get('inVocab'), union_of_None_type_or_booltype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'inVocab', str).makeError("the `inVocab` field is not valid because:\n"+str(e))) else: self.inVocab = None try: self.symbols = load_field(doc.get('symbols'), uri_array_of_strtype_True_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'symbols', str).makeError("the `symbols` field is not valid because:\n"+str(e))) try: self.type = load_field(doc.get('type'), typedsl_enum_d961d79c225752b9fadb617367615ab176b47d77Loader_2, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'type', str).makeError("the `type` field is not valid because:\n"+str(e))) if 'doc' in doc: try: self.doc = load_field(doc.get('doc'), union_of_None_type_or_strtype_or_array_of_strtype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'doc', str).makeError("the `doc` field is not valid because:\n"+str(e))) else: self.doc = None if 'docParent' in doc: try: self.docParent = load_field(doc.get('docParent'), uri_union_of_None_type_or_strtype_False_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'docParent', str).makeError("the `docParent` field is not valid because:\n"+str(e))) else: self.docParent = None if 'docChild' in doc: try: self.docChild = load_field(doc.get('docChild'), uri_union_of_None_type_or_strtype_or_array_of_strtype_False_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'docChild', str).makeError("the `docChild` field is not valid because:\n"+str(e))) else: self.docChild = None if 'docAfter' in doc: try: self.docAfter = load_field(doc.get('docAfter'), uri_union_of_None_type_or_strtype_False_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'docAfter', str).makeError("the `docAfter` field is not valid because:\n"+str(e))) else: self.docAfter = None if 'jsonldPredicate' in doc: try: self.jsonldPredicate = load_field(doc.get('jsonldPredicate'), union_of_None_type_or_strtype_or_JsonldPredicateLoader, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'jsonldPredicate', str).makeError("the `jsonldPredicate` field is not valid because:\n"+str(e))) else: self.jsonldPredicate = None if 'documentRoot' in doc: try: self.documentRoot = load_field(doc.get('documentRoot'), union_of_None_type_or_booltype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'documentRoot', str).makeError("the `documentRoot` field is not valid because:\n"+str(e))) else: self.documentRoot = None if 'extends' in doc: try: self.extends = load_field(doc.get('extends'), uri_union_of_None_type_or_strtype_or_array_of_strtype_False_False_1, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'extends', str).makeError("the `extends` field is not valid because:\n"+str(e))) else: self.extends = None self.extension_fields = {} for k in doc.keys(): if k not in self.attrs: if ":" in k: ex = expand_url(k, u"", loadingOptions, scoped_id=False, vocab_term=False) self.extension_fields[ex] = doc[k] else: errors.append(SourceLine(doc, k, str).makeError("invalid field `%s`, expected one of: `name`, `inVocab`, `symbols`, `type`, `doc`, `docParent`, `docChild`, `docAfter`, `jsonldPredicate`, `documentRoot`, `extends`" % (k))) break if errors: raise ValidationException("Trying 'SaladEnumSchema'\n"+"\n".join(errors)) def save(self, top=False, base_url="", relative_uris=True): r = {} for ef in self.extension_fields: r[prefix_url(ef, self.loadingOptions.vocab)] = self.extension_fields[ef] if self.name is not None: u = save_relative_uri(self.name, base_url, True, None, relative_uris) if u: r['name'] = u if self.inVocab is not None: r['inVocab'] = save(self.inVocab, top=False, base_url=self.name, relative_uris=relative_uris) if self.symbols is not None: u = save_relative_uri(self.symbols, self.name, True, None, relative_uris) if u: r['symbols'] = u if self.type is not None: r['type'] = save(self.type, top=False, base_url=self.name, relative_uris=relative_uris) if self.doc is not None: r['doc'] = save(self.doc, top=False, base_url=self.name, relative_uris=relative_uris) if self.docParent is not None: u = save_relative_uri(self.docParent, self.name, False, None, relative_uris) if u: r['docParent'] = u if self.docChild is not None: u = save_relative_uri(self.docChild, self.name, False, None, relative_uris) if u: r['docChild'] = u if self.docAfter is not None: u = save_relative_uri(self.docAfter, self.name, False, None, relative_uris) if u: r['docAfter'] = u if self.jsonldPredicate is not None: r['jsonldPredicate'] = save(self.jsonldPredicate, top=False, base_url=self.name, relative_uris=relative_uris) if self.documentRoot is not None: r['documentRoot'] = save(self.documentRoot, top=False, base_url=self.name, relative_uris=relative_uris) if self.extends is not None: u = save_relative_uri(self.extends, self.name, False, 1, relative_uris) if u: r['extends'] = u if top and self.loadingOptions.namespaces: r["$namespaces"] = self.loadingOptions.namespaces return r attrs = frozenset(['name', 'inVocab', 'symbols', 'type', 'doc', 'docParent', 'docChild', 'docAfter', 'jsonldPredicate', 'documentRoot', 'extends']) class Documentation(NamedType, DocType): """ A documentation section. This type exists to facilitate self-documenting schemas but has no role in formal validation. """ def __init__(self, _doc, baseuri, loadingOptions, docRoot=None): doc = copy.copy(_doc) if hasattr(_doc, 'lc'): doc.lc.data = _doc.lc.data doc.lc.filename = _doc.lc.filename errors = [] self.loadingOptions = loadingOptions if 'name' in doc: try: self.name = load_field(doc.get('name'), uri_strtype_True_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'name', str).makeError("the `name` field is not valid because:\n"+str(e))) else: self.name = None if self.name is None: if docRoot is not None: self.name = docRoot else: raise ValidationException("Missing name") baseuri = self.name if 'inVocab' in doc: try: self.inVocab = load_field(doc.get('inVocab'), union_of_None_type_or_booltype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'inVocab', str).makeError("the `inVocab` field is not valid because:\n"+str(e))) else: self.inVocab = None if 'doc' in doc: try: self.doc = load_field(doc.get('doc'), union_of_None_type_or_strtype_or_array_of_strtype, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'doc', str).makeError("the `doc` field is not valid because:\n"+str(e))) else: self.doc = None if 'docParent' in doc: try: self.docParent = load_field(doc.get('docParent'), uri_union_of_None_type_or_strtype_False_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'docParent', str).makeError("the `docParent` field is not valid because:\n"+str(e))) else: self.docParent = None if 'docChild' in doc: try: self.docChild = load_field(doc.get('docChild'), uri_union_of_None_type_or_strtype_or_array_of_strtype_False_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'docChild', str).makeError("the `docChild` field is not valid because:\n"+str(e))) else: self.docChild = None if 'docAfter' in doc: try: self.docAfter = load_field(doc.get('docAfter'), uri_union_of_None_type_or_strtype_False_False_None, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'docAfter', str).makeError("the `docAfter` field is not valid because:\n"+str(e))) else: self.docAfter = None try: self.type = load_field(doc.get('type'), typedsl_enum_056429f0e9355680bd9b2411dc96a69c7ff2e76bLoader_2, baseuri, loadingOptions) except ValidationException as e: errors.append(SourceLine(doc, 'type', str).makeError("the `type` field is not valid because:\n"+str(e))) self.extension_fields = {} for k in doc.keys(): if k not in self.attrs: if ":" in k: ex = expand_url(k, u"", loadingOptions, scoped_id=False, vocab_term=False) self.extension_fields[ex] = doc[k] else: errors.append(SourceLine(doc, k, str).makeError("invalid field `%s`, expected one of: `name`, `inVocab`, `doc`, `docParent`, `docChild`, `docAfter`, `type`" % (k))) break if errors: raise ValidationException("Trying 'Documentation'\n"+"\n".join(errors)) def save(self, top=False, base_url="", relative_uris=True): r = {} for ef in self.extension_fields: r[prefix_url(ef, self.loadingOptions.vocab)] = self.extension_fields[ef] if self.name is not None: u = save_relative_uri(self.name, base_url, True, None, relative_uris) if u: r['name'] = u if self.inVocab is not None: r['inVocab'] = save(self.inVocab, top=False, base_url=self.name, relative_uris=relative_uris) if self.doc is not None: r['doc'] = save(self.doc, top=False, base_url=self.name, relative_uris=relative_uris) if self.docParent is not None: u = save_relative_uri(self.docParent, self.name, False, None, relative_uris) if u: r['docParent'] = u if self.docChild is not None: u = save_relative_uri(self.docChild, self.name, False, None, relative_uris) if u: r['docChild'] = u if self.docAfter is not None: u = save_relative_uri(self.docAfter, self.name, False, None, relative_uris) if u: r['docAfter'] = u if self.type is not None: r['type'] = save(self.type, top=False, base_url=self.name, relative_uris=relative_uris) if top and self.loadingOptions.namespaces: r["$namespaces"] = self.loadingOptions.namespaces return r attrs = frozenset(['name', 'inVocab', 'doc', 'docParent', 'docChild', 'docAfter', 'type']) _vocab = { "Any": "https://w3id.org/cwl/salad#Any", "ArraySchema": "https://w3id.org/cwl/salad#ArraySchema", "DocType": "https://w3id.org/cwl/salad#DocType", "Documentation": "https://w3id.org/cwl/salad#Documentation", "Documented": "https://w3id.org/cwl/salad#Documented", "EnumSchema": "https://w3id.org/cwl/salad#EnumSchema", "JsonldPredicate": "https://w3id.org/cwl/salad#JsonldPredicate", "NamedType": "https://w3id.org/cwl/salad#NamedType", "PrimitiveType": "https://w3id.org/cwl/salad#PrimitiveType", "RecordField": "https://w3id.org/cwl/salad#RecordField", "RecordSchema": "https://w3id.org/cwl/salad#RecordSchema", "SaladEnumSchema": "https://w3id.org/cwl/salad#SaladEnumSchema", "SaladRecordField": "https://w3id.org/cwl/salad#SaladRecordField", "SaladRecordSchema": "https://w3id.org/cwl/salad#SaladRecordSchema", "SchemaDefinedType": "https://w3id.org/cwl/salad#SchemaDefinedType", "SpecializeDef": "https://w3id.org/cwl/salad#SpecializeDef", "array": "https://w3id.org/cwl/salad#array", "boolean": "http://www.w3.org/2001/XMLSchema#boolean", "documentation": "https://w3id.org/cwl/salad#documentation", "double": "http://www.w3.org/2001/XMLSchema#double", "enum": "https://w3id.org/cwl/salad#enum", "float": "http://www.w3.org/2001/XMLSchema#float", "int": "http://www.w3.org/2001/XMLSchema#int", "long": "http://www.w3.org/2001/XMLSchema#long", "null": "https://w3id.org/cwl/salad#null", "record": "https://w3id.org/cwl/salad#record", "string": "http://www.w3.org/2001/XMLSchema#string", } _rvocab = { "https://w3id.org/cwl/salad#Any": "Any", "https://w3id.org/cwl/salad#ArraySchema": "ArraySchema", "https://w3id.org/cwl/salad#DocType": "DocType", "https://w3id.org/cwl/salad#Documentation": "Documentation", "https://w3id.org/cwl/salad#Documented": "Documented", "https://w3id.org/cwl/salad#EnumSchema": "EnumSchema", "https://w3id.org/cwl/salad#JsonldPredicate": "JsonldPredicate", "https://w3id.org/cwl/salad#NamedType": "NamedType", "https://w3id.org/cwl/salad#PrimitiveType": "PrimitiveType", "https://w3id.org/cwl/salad#RecordField": "RecordField", "https://w3id.org/cwl/salad#RecordSchema": "RecordSchema", "https://w3id.org/cwl/salad#SaladEnumSchema": "SaladEnumSchema", "https://w3id.org/cwl/salad#SaladRecordField": "SaladRecordField", "https://w3id.org/cwl/salad#SaladRecordSchema": "SaladRecordSchema", "https://w3id.org/cwl/salad#SchemaDefinedType": "SchemaDefinedType", "https://w3id.org/cwl/salad#SpecializeDef": "SpecializeDef", "https://w3id.org/cwl/salad#array": "array", "http://www.w3.org/2001/XMLSchema#boolean": "boolean", "https://w3id.org/cwl/salad#documentation": "documentation", "http://www.w3.org/2001/XMLSchema#double": "double", "https://w3id.org/cwl/salad#enum": "enum", "http://www.w3.org/2001/XMLSchema#float": "float", "http://www.w3.org/2001/XMLSchema#int": "int", "http://www.w3.org/2001/XMLSchema#long": "long", "https://w3id.org/cwl/salad#null": "null", "https://w3id.org/cwl/salad#record": "record", "http://www.w3.org/2001/XMLSchema#string": "string", } strtype = _PrimitiveLoader((str, six.text_type)) inttype = _PrimitiveLoader(int) floattype = _PrimitiveLoader(float) booltype = _PrimitiveLoader(bool) None_type = _PrimitiveLoader(type(None)) Any_type = _AnyLoader() DocumentedLoader = _RecordLoader(Documented) PrimitiveTypeLoader = _EnumLoader(("null", "boolean", "int", "long", "float", "double", "string",)) AnyLoader = _EnumLoader(("Any",)) RecordFieldLoader = _RecordLoader(RecordField) RecordSchemaLoader = _RecordLoader(RecordSchema) EnumSchemaLoader = _RecordLoader(EnumSchema) ArraySchemaLoader = _RecordLoader(ArraySchema) JsonldPredicateLoader = _RecordLoader(JsonldPredicate) SpecializeDefLoader = _RecordLoader(SpecializeDef) NamedTypeLoader = _RecordLoader(NamedType) DocTypeLoader = _RecordLoader(DocType) SchemaDefinedTypeLoader = _RecordLoader(SchemaDefinedType) SaladRecordFieldLoader = _RecordLoader(SaladRecordField) SaladRecordSchemaLoader = _RecordLoader(SaladRecordSchema) SaladEnumSchemaLoader = _RecordLoader(SaladEnumSchema) DocumentationLoader = _RecordLoader(Documentation) array_of_strtype = _ArrayLoader(strtype) union_of_None_type_or_strtype_or_array_of_strtype = _UnionLoader((None_type, strtype, array_of_strtype,)) uri_strtype_True_False_None = _URILoader(strtype, True, False, None) union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype = _UnionLoader((PrimitiveTypeLoader, RecordSchemaLoader, EnumSchemaLoader, ArraySchemaLoader, strtype,)) array_of_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype = _ArrayLoader(union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype) union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_or_array_of_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype = _UnionLoader((PrimitiveTypeLoader, RecordSchemaLoader, EnumSchemaLoader, ArraySchemaLoader, strtype, array_of_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype,)) typedsl_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_or_array_of_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_2 = _TypeDSLLoader(union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_or_array_of_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype, 2) array_of_RecordFieldLoader = _ArrayLoader(RecordFieldLoader) union_of_None_type_or_array_of_RecordFieldLoader = _UnionLoader((None_type, array_of_RecordFieldLoader,)) idmap_fields_union_of_None_type_or_array_of_RecordFieldLoader = _IdMapLoader(union_of_None_type_or_array_of_RecordFieldLoader, 'name', 'type') enum_d9cba076fca539106791a4f46d198c7fcfbdb779Loader = _EnumLoader(("record",)) typedsl_enum_d9cba076fca539106791a4f46d198c7fcfbdb779Loader_2 = _TypeDSLLoader(enum_d9cba076fca539106791a4f46d198c7fcfbdb779Loader, 2) uri_array_of_strtype_True_False_None = _URILoader(array_of_strtype, True, False, None) enum_d961d79c225752b9fadb617367615ab176b47d77Loader = _EnumLoader(("enum",)) typedsl_enum_d961d79c225752b9fadb617367615ab176b47d77Loader_2 = _TypeDSLLoader(enum_d961d79c225752b9fadb617367615ab176b47d77Loader, 2) uri_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_or_array_of_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_False_True_2 = _URILoader(union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype_or_array_of_union_of_PrimitiveTypeLoader_or_RecordSchemaLoader_or_EnumSchemaLoader_or_ArraySchemaLoader_or_strtype, False, True, 2) enum_d062602be0b4b8fd33e69e29a841317b6ab665bcLoader = _EnumLoader(("array",)) typedsl_enum_d062602be0b4b8fd33e69e29a841317b6ab665bcLoader_2 = _TypeDSLLoader(enum_d062602be0b4b8fd33e69e29a841317b6ab665bcLoader, 2) union_of_None_type_or_strtype = _UnionLoader((None_type, strtype,)) uri_union_of_None_type_or_strtype_True_False_None = _URILoader(union_of_None_type_or_strtype, True, False, None) union_of_None_type_or_booltype = _UnionLoader((None_type, booltype,)) union_of_None_type_or_inttype = _UnionLoader((None_type, inttype,)) uri_strtype_False_False_1 = _URILoader(strtype, False, False, 1) uri_union_of_None_type_or_strtype_False_False_None = _URILoader(union_of_None_type_or_strtype, False, False, None) uri_union_of_None_type_or_strtype_or_array_of_strtype_False_False_None = _URILoader(union_of_None_type_or_strtype_or_array_of_strtype, False, False, None) union_of_None_type_or_strtype_or_JsonldPredicateLoader = _UnionLoader((None_type, strtype, JsonldPredicateLoader,)) union_of_None_type_or_Any_type = _UnionLoader((None_type, Any_type,)) array_of_SaladRecordFieldLoader = _ArrayLoader(SaladRecordFieldLoader) union_of_None_type_or_array_of_SaladRecordFieldLoader = _UnionLoader((None_type, array_of_SaladRecordFieldLoader,)) idmap_fields_union_of_None_type_or_array_of_SaladRecordFieldLoader = _IdMapLoader(union_of_None_type_or_array_of_SaladRecordFieldLoader, 'name', 'type') uri_union_of_None_type_or_strtype_or_array_of_strtype_False_False_1 = _URILoader(union_of_None_type_or_strtype_or_array_of_strtype, False, False, 1) array_of_SpecializeDefLoader = _ArrayLoader(SpecializeDefLoader) union_of_None_type_or_array_of_SpecializeDefLoader = _UnionLoader((None_type, array_of_SpecializeDefLoader,)) idmap_specialize_union_of_None_type_or_array_of_SpecializeDefLoader = _IdMapLoader(union_of_None_type_or_array_of_SpecializeDefLoader, 'specializeFrom', 'specializeTo') enum_056429f0e9355680bd9b2411dc96a69c7ff2e76bLoader = _EnumLoader(("documentation",)) typedsl_enum_056429f0e9355680bd9b2411dc96a69c7ff2e76bLoader_2 = _TypeDSLLoader(enum_056429f0e9355680bd9b2411dc96a69c7ff2e76bLoader, 2) union_of_SaladRecordSchemaLoader_or_SaladEnumSchemaLoader_or_DocumentationLoader = _UnionLoader((SaladRecordSchemaLoader, SaladEnumSchemaLoader, DocumentationLoader,)) array_of_union_of_SaladRecordSchemaLoader_or_SaladEnumSchemaLoader_or_DocumentationLoader = _ArrayLoader(union_of_SaladRecordSchemaLoader_or_SaladEnumSchemaLoader_or_DocumentationLoader) union_of_SaladRecordSchemaLoader_or_SaladEnumSchemaLoader_or_DocumentationLoader_or_array_of_union_of_SaladRecordSchemaLoader_or_SaladEnumSchemaLoader_or_DocumentationLoader = _UnionLoader((SaladRecordSchemaLoader, SaladEnumSchemaLoader, DocumentationLoader, array_of_union_of_SaladRecordSchemaLoader_or_SaladEnumSchemaLoader_or_DocumentationLoader,)) def load_document(doc, baseuri=None, loadingOptions=None): if baseuri is None: baseuri = file_uri(os.getcwd()) + "/" if loadingOptions is None: loadingOptions = LoadingOptions() return _document_load(union_of_SaladRecordSchemaLoader_or_SaladEnumSchemaLoader_or_DocumentationLoader_or_array_of_union_of_SaladRecordSchemaLoader_or_SaladEnumSchemaLoader_or_DocumentationLoader, doc, baseuri, loadingOptions)