-
Notifications
You must be signed in to change notification settings - Fork 0
/
graph.py
91 lines (81 loc) · 3.54 KB
/
graph.py
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
"""Knowledge Graph Parameter Type."""
from typing import Optional, Set, List, Any
from cmem.cmempy.dp.proxy.graph import get_graphs_list
from cmem_plugin_base.dataintegration.context import PluginContext
from cmem_plugin_base.dataintegration.types import StringParameterType, Autocompletion
from cmem_plugin_base.dataintegration.utils import setup_cmempy_user_access
class GraphParameterType(StringParameterType):
"""Knowledge Graph parameter type."""
allow_only_autocompleted_values: bool = False
autocomplete_value_with_labels: bool = True
classes: Optional[Set[str]] = None
def __init__(
self,
show_di_graphs: bool = False,
show_system_graphs: bool = False,
show_graphs_without_class: bool = False,
classes: Optional[List[str]] = None,
allow_only_autocompleted_values: bool = True,
):
"""
Knowledge Graph parameter type.
:param show_di_graphs: show DI project graphs
:param show_system_graphs: show system graphs such as shape and query catalogs
:param classes: allowed classes of the shown graphs
- if None -> defaults to di:Dataset and void:Dataset
:param allow_only_autocompleted_values: allow entering new graph URLs
"""
self.show_di_graphs = show_di_graphs
self.show_system_graphs = show_system_graphs
self.show_graphs_without_class = show_graphs_without_class
self.allow_only_autocompleted_values = allow_only_autocompleted_values
if classes:
self.classes = set(classes)
else:
self.classes = {
"https://vocab.eccenca.com/di/Dataset",
"http://rdfs.org/ns/void#Dataset",
}
def autocomplete(
self,
query_terms: list[str],
depend_on_parameter_values: list[Any],
context: PluginContext,
) -> list[Autocompletion]:
setup_cmempy_user_access(context=context.user)
graphs = get_graphs_list()
result = []
for _ in graphs:
iri = _["iri"]
title = _["label"]["title"]
label = f"{title} ({iri})"
assigned_classes = set(_["assignedClasses"])
# ignore DI project graphs
if self.show_di_graphs is False and _["diProjectGraph"] is True:
continue
# ignore system resource graphs
if self.show_system_graphs is False and _["systemResource"] is True:
continue
# show graphs without assigned classes only if explicitly wanted
if len(assigned_classes) == 0:
if self.show_graphs_without_class is True:
result.append(Autocompletion(value=iri, label=label))
continue
# ignore graphs which do not match the requested classes
if (
self.classes is not None
and len(assigned_classes) > 0
and len(self.classes.intersection(assigned_classes)) == 0
):
continue
# if no search terms are given: add all remaining graphs to list
if len(query_terms) == 0:
result.append(Autocompletion(value=iri, label=label))
continue
# show only graphs which match the given terms
for term in query_terms:
if term.lower() in label.lower():
result.append(Autocompletion(value=iri, label=label))
continue
result.sort(key=lambda x: x.label) # type: ignore
return list(set(result))