ENH: "Do not retrieve" option for json schema exporter.

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)
 

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")