Skip to content

kg_schema

KGSchema

Stores a Graph object and some metadata corresponding to a KG schema

Source code in exe_kg_lib/classes/kg_schema.py
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
class KGSchema:
    """
    Stores a Graph object and some metadata corresponding to a KG schema
    """

    def __init__(
        self,
        path: str,
        shacl_shapes_path: str,
        generated_schema_path: str,
        generated_shacl_shapes_path: str,
        namespace: str,
        namespace_prefix: str,
    ):
        self.path = path  # path of the main KG schema definition, can be local or remote
        self.generated_schema_path = (
            generated_schema_path  # path of file containing generated schema for this schema, can be local or remote
        )
        self.namespace = Namespace(namespace)
        self.namespace_prefix = namespace_prefix

        self.kg = Graph(bind_namespaces="rdflib")
        self.kg.parse(self.path, format="n3")

        self.generated_schema_kg = Graph(bind_namespaces="rdflib")
        if self.generated_schema_path:
            self.generated_schema_kg.parse(self.generated_schema_path, format="n3")

        # shacl
        self.shacl_shapes_path = shacl_shapes_path  # path of file containing main shacl shapes, can be local or remote
        self.generated_shacl_shapes_path = (
            generated_shacl_shapes_path  # path of file containing generated shacl shapes, can be local or remote
        )

        self.shacl_shapes_s = self.read_shacl_shapes(self.shacl_shapes_path)  # shacl shapes are stored as string

        if self.generated_shacl_shapes_path:
            self.shacl_shapes_s += self.read_shacl_shapes(self.generated_shacl_shapes_path)

    @classmethod
    def from_schema_info(cls, schema_info: Dict[str, str]):
        return cls(
            schema_info["path"],
            schema_info["shacl_shapes_path"],
            schema_info["generated_schema_path"],
            schema_info["generated_shacl_shapes_path"],
            schema_info["namespace"],
            schema_info["namespace_prefix"],
        )

    @staticmethod
    def read_shacl_shapes(path: str):
        if path.startswith("http"):
            return requests.get(path).text
        else:
            with open(path) as f:
                return f.read()