Merge pull request #64 from OpenSemanticLab/add-support-for-characteristics

Add basic support for characteristics

Author: simontaurus

src/osw/core.py

@@ -356,20 +356,26 @@ def _fetch_schema(self, fetchSchemaParam: _FetchSchemaParam = None) -> None:
         if not page.exists:
             print(f"Error: Page {schema_title} does not exist")
             return
-        if schema_title.startswith("Category:"):
+        # not only in the JsonSchema namespace the schema is located in the main sot
+        # in all other namespaces, the json_schema slot is used
+        if schema_title.startswith("JsonSchema:"):
+            schema_str = ""
+            if page.get_slot_content("main"):
+                schema_str = json.dumps(page.get_slot_content("main"))
+        else:
             schema_str = ""
             if page.get_slot_content("jsonschema"):
                 schema_str = json.dumps(page.get_slot_content("jsonschema"))
-        else:
-            schema_str = page.get_slot_content("main")
-            if schema_str and not isinstance(schema_str, str):
-                schema_str = json.dumps(schema_str)
         if (schema_str is None) or (schema_str == ""):
             print(f"Error: Schema {schema_title} does not exist")
-            return
+            schema_str = "{}"  # empty schema to make reference work
         schema = json.loads(
-            schema_str.replace("$ref", "dollarref")
-        )  # '$' is a special char for root object in jsonpath
+            schema_str.replace(
+                "$ref", "dollarref"
+            ).replace(  # '$' is a special char for root object in jsonpath
+                '"allOf": [', '"allOf": [{},'
+            )  # fix https://github.com/koxudaxi/datamodel-code-generator/issues/1910
+        )
         print(f"Fetch {schema_title}")
 
         jsonpath_expr = parse("$..dollarref")
@@ -626,15 +632,15 @@ def load_entity(
                 print("Error: no schema defined")
 
             elif len(schemas) == 1:
-                cls = schemas[0]["title"]
-                entity: model.Entity = eval(f"model.{cls}(**jsondata)")
+                cls = getattr(model, schemas[0]["title"])
+                entity: model.Entity = cls(**jsondata)
 
             else:
                 bases = []
                 for schema in schemas:
-                    bases.append(eval("model." + schema["title"]))
+                    bases.append(getattr(model, schema["title"]))
                 cls = create_model("Test", __base__=tuple(bases))
-                entity = cls(**jsondata)
+                entity: model.Entity = cls(**jsondata)
 
             if entity is not None:
                 # make sure we do not override existing meta data
@@ -957,6 +963,8 @@ def store_entity(
         if not isinstance(param.entities, list):
             param.entities = [param.entities]
 
+        param: OSW.StoreEntityParam = param
+
         max_index = len(param.entities)
 
         meta_category = self.site.get_page(
@@ -970,12 +978,12 @@ def store_entity(
                 meta_category = self.site.get_page(
                     WtSite.GetPageParam(titles=[param.meta_category_title])
                 ).pages[0]
-                param.meta_category_template_str = meta_category.get_slot_content(
+                meta_category_template_str = meta_category.get_slot_content(
                     "schema_template"
                 )
-            if param.meta_category_template_str:
+            if meta_category_template_str:
                 meta_category_template = compile_handlebars_template(
-                    param.meta_category_template_str
+                    meta_category_template_str
                 )
 
         def store_entity_(
@@ -1010,10 +1018,26 @@ def store_entity_(
                     schema_str = eval_compiled_handlebars_template(
                         meta_category_template,
                         page.get_slot_content("jsondata"),
-                        {"_page_title": entity_title},
+                        {
+                            "_page_title": entity_title,  # legacy
+                            "_current_subject_": entity_title,
+                        },
                     )
                     schema = json.loads(schema_str)
-                    page.set_slot_content("jsonschema", schema)
+                    # put generated schema in definitions section
+                    # currently only enabled for Characteristics
+                    if hasattr(model, "MetaCharacteristic") and isinstance(
+                        entity, model.MetaCharacteristic
+                    ):
+                        new_schema = {
+                            "$defs": {"generated": schema},
+                            "allOf": [{"$ref": "#/$defs/generated"}],
+                        }
+                        new_schema["@context"] = schema.pop("@context", None)
+                        new_schema["title"] = schema.pop("title", "")
+                        schema["title"] = "Generated" + new_schema["title"]
+                        schema = new_schema
+                    page.set_slot_content("jsonschema", new_schema)
                 except Exception as e:
                     print(
                         f"Schema generation from template failed for " f"{entity}: {e}"

src/osw/utils/wiki.py

@@ -1,7 +1,8 @@
-from typing import Union
+from typing import Type, Union
 from uuid import UUID
 
 import osw.model.entity as model
+from osw.model.static import OswBaseModel
 
 
 def get_osw_id(uuid: UUID) -> str:
@@ -35,7 +36,7 @@ def get_uuid(osw_id) -> UUID:
     return UUID(osw_id.replace("OSW", ""))
 
 
-def get_namespace(entity: model.Entity) -> Union[str, None]:
+def get_namespace(entity: Union[OswBaseModel, Type[OswBaseModel]]) -> Union[str, None]:
     """determines the wiki namespace based on the entity's type/class
 
     Parameters
@@ -54,6 +55,7 @@ def get_namespace(entity: model.Entity) -> Union[str, None]:
             namespace = entity.meta.wiki_page.namespace
 
     #  (model classes may not exist => try except)
+    #  note: this may not work properly with dynamic reloaded model module
     if namespace is None:
         try:
             if issubclass(entity, model.Entity):
@@ -62,14 +64,20 @@ def get_namespace(entity: model.Entity) -> Union[str, None]:
             pass
     if namespace is None:
         try:
-            if isinstance(entity, model.Item):
-                namespace = "Item"
+            if isinstance(entity, model.Category):
+                namespace = "Category"
         except AttributeError:
             pass
     if namespace is None:
         try:
-            if isinstance(entity, model.Category):
+            if issubclass(entity, model.Characteristic):
                 namespace = "Category"
+        except (TypeError, AttributeError):
+            pass
+    if namespace is None:
+        try:
+            if isinstance(entity, model.Item):
+                namespace = "Item"
         except AttributeError:
             pass
     if namespace is None:

tests/integration/store_and_load_test.py

@@ -171,3 +171,89 @@ def test_statement_creation(wiki_domain, wiki_username, wiki_password):
     assert f"Item:{OSW.get_osw_id(my_entity.uuid)}" in full_page_titles
 
     osw.delete_entity(my_entity)
+
+
+def test_characteristic_creation(wiki_domain, wiki_username, wiki_password):
+    cm = CredentialManager()
+    cm.add_credential(
+        CredentialManager.UserPwdCredential(
+            iri=wiki_domain, username=wiki_username, password=wiki_password
+        )
+    )
+    wtsite = WtSite(WtSite.WtSiteConfig(iri=wiki_domain, cred_mngr=cm))
+    osw = OSW(site=wtsite)
+
+    osw.fetch_schema(
+        OSW.FetchSchemaParam(
+            schema_title=[
+                "Category:OSWffe74f291d354037b318c422591c5023",  # MetaCharacteristic
+                "Category:Property",
+            ],
+            mode="replace",
+        )
+    )
+
+    # my_property = model.Property
+
+    # Create a characteristic as instance of MetaCharacteristic
+    my_characteristic = model.MetaCharacteristic(
+        uuid="efad8086-4a76-47ca-b278-f5f944e5754b",
+        name="TestCharacteristic",
+        label=[model.Label(text="Test Characteristic")],
+        properties=[
+            model.PrimitiveProperty(
+                uuid="766e7171-a183-4f9c-a9af-28cfd27fb1d9",
+                name="test_property",
+                type="string",
+                property_type="SimpleProperty",
+            ),
+            model.PrimitiveProperty(
+                uuid="766e7171-a183-4f9c-a9af-28cfd27fb1d1",
+                name="test_property2",
+                rdf_property="Property:TestPropertyWithSchema",
+                property_type="SimpleProperty",
+            ),
+        ],
+    )
+
+    # store it as instance, which generates the schema
+    # note: namespace and meta_category should be detected automatically in the future
+    osw.store_entity(
+        OSW.StoreEntityParam(
+            entities=[my_characteristic],
+            namespace="Category",
+            meta_category_title="Category:OSWffe74f291d354037b318c422591c5023",
+            overwrite=OverwriteOptions.true,
+        )
+    )
+
+    pages = osw.site.get_page(
+        WtSite.GetPageParam(titles=["Property:TestPropertyWithSchema"])
+    ).pages
+    pp = pages[0]
+    # schema: dict = pp.get_slot_content("jsonschema")
+    new_schema = {"type": "number", "description": "Imported from property schema"}
+    pp.set_slot_content("jsonschema", new_schema)
+    pp.edit()
+
+    # load the characteristic as category
+    osw.fetch_schema(
+        OSW.FetchSchemaParam(
+            schema_title=[
+                "Category:OSWefad80864a7647cab278f5f944e5754b"  # TestCharacteristic
+            ],
+            mode="append",
+        )
+    )
+
+    # create an instance of the characteristic
+    t = model.TestCharacteristic(test_property="Test", test_property2=1)
+    assert t.test_property == "Test"
+    assert t.test_property2 == 1
+
+    # cleanup (disabled for paralle test matrix)
+    # my_characteristic.meta = model.Meta(
+    #     wiki_page=model.WikiPage(namespace="Category")
+    # )  # namespace detection fails otherwise
+    # osw.delete_entity(my_characteristic)
+    # pp.delete()