1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161
|
"""Generate langauge specific loaders for a particular SALAD schema."""
import sys
from io import TextIOWrapper
from typing import (
Any,
Dict,
List,
MutableMapping,
MutableSequence,
Optional,
TextIO,
Union,
)
from . import schema
from .codegen_base import CodeGenBase
from .exceptions import SchemaSaladException
from .java_codegen import JavaCodeGen
from .python_codegen import PythonCodeGen
from .ref_resolver import Loader
from .schema import shortname
from .utils import aslist
def codegen(
lang: str,
i: List[Dict[str, str]],
schema_metadata: Dict[str, Any],
loader: Loader,
target: Optional[str] = None,
examples: Optional[str] = None,
) -> None:
"""Generate classes with loaders for the given Schema Salad description."""
j = schema.extend_and_specialize(i, loader)
gen = None # type: Optional[CodeGenBase]
if lang == "python":
if target:
dest: Union[TextIOWrapper, TextIO] = open(
target, mode="w", encoding="utf-8"
)
else:
dest = sys.stdout
gen = PythonCodeGen(dest)
elif lang == "java":
gen = JavaCodeGen(
schema_metadata.get("$base", schema_metadata.get("id")),
target=target,
examples=examples,
)
else:
raise SchemaSaladException(f"Unsupported code generation language '{lang}'")
assert gen is not None
gen.prologue()
document_roots = []
for rec in j:
if rec["type"] in ("enum", "record"):
gen.type_loader(rec)
gen.add_vocab(shortname(rec["name"]), rec["name"])
for rec in j:
if rec["type"] == "enum":
for symbol in rec["symbols"]:
gen.add_vocab(shortname(symbol), symbol)
if rec["type"] == "record":
if rec.get("documentRoot"):
document_roots.append(rec["name"])
field_names = []
optional_fields = set()
for field in rec.get("fields", []):
field_name = shortname(field["name"])
field_names.append(field_name)
tp = field["type"]
if (
isinstance(tp, MutableSequence)
and tp[0] == "https://w3id.org/cwl/salad#null"
):
optional_fields.add(field_name)
idfield = ""
for field in rec.get("fields", []):
if field.get("jsonldPredicate") == "@id":
idfield = field.get("name")
gen.begin_class(
rec["name"],
aslist(rec.get("extends", [])),
rec.get("doc", ""),
rec.get("abstract", False),
field_names,
idfield,
optional_fields,
)
gen.add_vocab(shortname(rec["name"]), rec["name"])
for field in rec.get("fields", []):
if field.get("jsonldPredicate") == "@id":
subscope = field.get("subscope")
fieldpred = field["name"]
optional = bool("https://w3id.org/cwl/salad#null" in field["type"])
uri_loader = gen.uri_loader(
gen.type_loader(field["type"]), True, False, None
)
gen.declare_id_field(
fieldpred, uri_loader, field.get("doc"), optional, subscope
)
break
for field in rec.get("fields", []):
optional = bool("https://w3id.org/cwl/salad#null" in field["type"])
type_loader = gen.type_loader(field["type"])
jld = field.get("jsonldPredicate")
fieldpred = field["name"]
subscope = None
if isinstance(jld, MutableMapping):
ref_scope = jld.get("refScope")
if jld.get("typeDSL"):
type_loader = gen.typedsl_loader(type_loader, ref_scope)
elif jld.get("secondaryFilesDSL"):
type_loader = gen.secondaryfilesdsl_loader(type_loader)
elif jld.get("subscope"):
subscope = jld.get("subscope")
elif jld.get("_type") == "@id":
type_loader = gen.uri_loader(
type_loader, jld.get("identity", False), False, ref_scope
)
elif jld.get("_type") == "@vocab":
type_loader = gen.uri_loader(
type_loader, False, True, ref_scope
)
map_subject = jld.get("mapSubject")
if map_subject:
type_loader = gen.idmap_loader(
field["name"],
type_loader,
map_subject,
jld.get("mapPredicate"),
)
if "_id" in jld and jld["_id"][0] != "@":
fieldpred = jld["_id"]
if jld == "@id":
continue
gen.declare_field(fieldpred, type_loader, field.get("doc"), optional)
gen.end_class(rec["name"], field_names)
root_type = list(document_roots)
root_type.append({"type": "array", "items": document_roots})
gen.epilogue(gen.type_loader(root_type))
|