diff --git a/src/caosadvancedtools/json_schema_exporter.py b/src/caosadvancedtools/json_schema_exporter.py
index 172cc1c089c56d092a61feb705e15c6cc1aa2e83..7760626d97300c336a05499bf582aa0fa4c6959c 100644
--- a/src/caosadvancedtools/json_schema_exporter.py
+++ b/src/caosadvancedtools/json_schema_exporter.py
@@ -40,6 +40,8 @@ class JsonSchemaExporter:
additional_options_for_text_props: dict = None,
units_in_description: bool = True,
do_not_create: List[str] = None,
+ do_not_retrieve: List[str] = None,
+ no_remote: bool = False,
):
"""Set up a JsonSchemaExporter, which can then be applied on RecordTypes.
@@ -63,17 +65,27 @@ class JsonSchemaExporter:
A list of RedcordType names, for which there should be no option
to create them. Instead, only the choice of existing elements should
be given.
+ do_not_retrieve : list[str]
+ A list of RedcordType names, for which no Records shall be retrieved. Instead, only an
+ object description should be given. If this list overlaps with the `do_not_create`
+ parameter, the behavior is undefined.
+ no_remote : bool
+ If True, do not attempt to connect to a LinkAhead server at all. Default is False.
"""
if not additional_options_for_text_props:
additional_options_for_text_props = {}
if not do_not_create:
do_not_create = []
+ if not do_not_retrieve:
+ do_not_retrieve = []
self._additional_properties = additional_properties
self._name_and_description_in_properties = name_and_description_in_properties
self._additional_options_for_text_props = additional_options_for_text_props
self._units_in_description = units_in_description
self._do_not_create = do_not_create
+ self._do_not_retrieve = do_not_retrieve
+ self._no_remote = no_remote
@staticmethod
def _make_required_list(rt: db.RecordType):
@@ -151,22 +163,31 @@ class JsonSchemaExporter:
prop_name = prop.datatype
if isinstance(prop.datatype, db.Entity):
prop_name = prop.datatype.name
- values = self._retrieve_enum_values(f"RECORD '{prop_name}'")
+ if prop_name in self._do_not_retrieve:
+ values = []
+ else:
+ values = self._retrieve_enum_values(f"RECORD '{prop_name}'")
if prop_name in self._do_not_create:
# Only a simple list of values
json_prop["enum"] = values
else:
- rt = db.execute_query(f"FIND RECORDTYPE WITH name='{prop_name}'",
- unique=True)
+ if self._no_remote:
+ rt = prop.datatype
+ else:
+ rt = db.execute_query(f"FIND RECORDTYPE WITH name='{prop_name}'",
+ unique=True)
subschema = self._make_segment_from_recordtype(rt)
- subschema["title"] = "Create new"
- json_prop["oneOf"] = [
- {
- "title": "Existing entries",
- "enum": values,
- },
- subschema
- ]
+ if values:
+ subschema["title"] = "Create new"
+ json_prop["oneOf"] = [
+ {
+ "title": "Existing entries",
+ "enum": values,
+ },
+ subschema
+ ]
+ else:
+ json_prop = subschema
else:
raise ValueError(
@@ -203,8 +224,10 @@ class JsonSchemaExporter:
return prop
- @staticmethod
- def _retrieve_enum_values(role: str):
+ def _retrieve_enum_values(self, role: str):
+
+ if self._no_remote:
+ return []
possible_values = db.execute_query(f"SELECT name, id FROM {role}")
@@ -277,7 +300,10 @@ def recordtype_to_json_schema(rt: db.RecordType, additional_properties: bool = T
name_and_description_in_properties: bool = False,
additional_options_for_text_props: Optional[dict] = None,
units_in_description: bool = True,
- do_not_create: List[str] = None):
+ do_not_create: List[str] = None,
+ do_not_retrieve: List[str] = None,
+ no_remote: bool = False,
+ ):
"""Create a jsonschema from a given RecordType that can be used, e.g., to
validate a json specifying a record of the given type.
@@ -306,6 +332,12 @@ def recordtype_to_json_schema(rt: db.RecordType, additional_properties: bool = T
A list of RedcordType names, for which there should be no option
to create them. Instead, only the choice of existing elements should
be given.
+ do_not_retrieve : list[str]
+ A list of RedcordType names, for which no Records shall be retrieved. Instead, only an
+ object description should be given. If this list overlaps with the `do_not_create`
+ parameter, the behavior is undefined.
+ no_remote : bool
+ If True, do not attempt to connect to a LinkAhead server at all. Default is False.
Returns
-------
@@ -320,6 +352,8 @@ def recordtype_to_json_schema(rt: db.RecordType, additional_properties: bool = T
additional_options_for_text_props=additional_options_for_text_props,
units_in_description=units_in_description,
do_not_create=do_not_create,
+ do_not_retrieve=do_not_retrieve,
+ no_remote=no_remote,
)
return exporter.recordtype_to_json_schema(rt)
diff --git a/unittests/test_json_schema_exporter.py b/unittests/test_json_schema_exporter.py
index 18a375363c175dd6d2fd1736023ef29afa110bfd..102a8bffcaac66fbe33675406213ef23c47239a4 100644
--- a/unittests/test_json_schema_exporter.py
+++ b/unittests/test_json_schema_exporter.py
@@ -27,6 +27,8 @@ import json
import linkahead as db
import caosadvancedtools.json_schema_exporter as jsex
+from collections import OrderedDict
+
from jsonschema import FormatChecker, validate, ValidationError
from pytest import raises
from unittest.mock import Mock, patch
@@ -85,6 +87,8 @@ def _mock_execute_query(query_string, unique=False, **kwargs):
return all_files
else:
print(f"Query string: {query_string}")
+ if unique is True:
+ return db.Entity()
return db.Container()
@@ -773,3 +777,42 @@ RT5:
assert rt4_deep.get_parents()[0].name == "RT3"
rt5_deep = model.get_deep("RT5")
assert rt5_deep.get_parents()[0].name == "RT5"
+
+
+@patch("linkahead.execute_query", new=Mock(side_effect=_mock_execute_query))
+def test_empty_retrieve():
+ """Special case: ``do_not_retrieve`` is set, or the retrieve result is empty."""
+ model_str = """
+RT1:
+ description: Some text.
+RT2:
+ obligatory_properties:
+ RT1:
+# some_text:
+# datatype: TEXT
+NoRecords:
+ description: A RecordType without Records.
+ recommended_properties:
+ some_text:
+ datatype: TEXT
+RT3:
+ obligatory_properties:
+ NoRecords:
+ """
+ model = parse_model_from_string(model_str)
+ schema_default = rtjs(model.get_deep("RT2"))
+ assert "oneOf" in schema_default["properties"]["RT1"]
+ assert any([el.get("title") == "Existing entries" for el in
+ schema_default["properties"]["RT1"]["oneOf"]])
+
+ schema_noexist = rtjs(model.get_deep("RT3"))
+ assert schema_noexist["properties"]["NoRecords"].get("type") == "object"
+
+ schema_noexist_noremote = rtjs(model.get_deep("RT3"), no_remote=True)
+ assert schema_noexist_noremote["properties"]["NoRecords"].get("type") == "object"
+ assert (schema_noexist_noremote["properties"]["NoRecords"].get("properties")
+ == OrderedDict([('some_text', {'type': 'string'})]))
+
+ schema_noexist_noretrieve = rtjs(model.get_deep("RT2"), do_not_retrieve=["RT1"])
+ assert schema_noexist_noretrieve["properties"]["RT1"].get("type") == "object"
+ assert "some_date" in schema_noexist_noretrieve["properties"]["RT1"].get("properties")