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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
|
"""
Part of the BlackStar filesystem (bsfs) module.
A copy of the license is provided with the project.
Author: Matthias Baumgartner, 2022
"""
# imports
import typing
# bsfs imports
from bsfs import schema as bsc
# inner-module imports
from . import ast
# exports
__all__ : typing.Sequence[str] = (
'Filter',
)
## code ##
class Filter():
# schema to validate against.
schema: bsc.Schema
def __init__(self, schema: bsc.Schema):
self.schema = schema
def parse(self, node: ast.filter.FilterExpression, subject: bsc.types._Vertex):
# subject is a node type
if not isinstance(subject, bsc.Node):
raise errors.ConsistencyError(f'Expected a node, found {subject}')
# subject exists in the schema
if subject not in self.schema.nodes:
raise errors.ConsistencyError(f'Invalid node type {subject}')
# root expression is valid
self._parse(node, subject)
# all tests passed
return True
def _parse_numerical_expression(self, node: ast.filter.FilterExpression, subject: bsc.types._Vertex):
if isinstance(node, ast.filter.And):
return self._and(node, subject)
elif isinstance(node, ast.filter.Or):
return self._or(node, subject)
elif isinstance(node, ast.filter.LessThan):
return self._lessThan(node, subject)
elif isinstance(node, ast.filter.GreaterThan):
return self._greaterThan(node, subject)
elif isinstance(node, ast.filter.Equals):
return self._equals(node, subject, numerical=True)
else:
raise errors.ConsistencyError(f'Expected a numerical expression, found {node}')
def __branch(self, node: typing.Union[ast.filter.Any, ast.filter.And], subject: bsc.types._Vertex):
# subject is a node type
if not isinstance(subject, bsc.Node):
raise errors.ConsistencyError(f'Expected a node, found {subject}')
# subject exists in the schema
if subject not in self.schema.nodes:
raise errors.ConsistencyError(f'Invalid node type {subject}')
# predicate is valid
dom, rng = self._parse_predicate_expression(node.predicate)
# subject is a subtype of the predicate's domain
if not subject <= dom:
raise errors.ConsistencyError(f'Expected type {dom}, found {subject}')
# child expression is valid
self._parse_filter_expression(node.expr, rng)
def _any(self, node: ast.filter.Any, subject: bsc.types._Vertex):
return self.__branch(node, subject)
def _all(self, node: ast.filter.All, subject: bsc.types._Vertex):
return self.__branch(node, subject)
def __agg(self, node: typing.Union[ast.filter.And, ast.filter.Or], subject: bsc.types._Vertex):
for expr in node:
# child expression is valid
self._parse_filter_expression(expr, subject)
def _and(self, node: ast.filter.And, subject: bsc.types._Vertex):
return self.__agg(node, subject)
def _or(self, node: ast.filter.Or, subject: bsc.types._Vertex):
return self.__agg(node, subject)
def _not(self, node: ast.filter.Not, subject: bsc.types._Vertex):
# child expression is valid
self._parse_filter_expression(node.expr, subject)
def _has(self, node: ast.filter.Has, subject: bsc.types._Vertex):
# subject is a node type
if not isinstance(subject, bsc.Node):
raise errors.ConsistencyError(f'Expected a node, found {subject}')
# subject exists in the schema
if subject not in self.schema.nodes:
raise errors.ConsistencyError(f'Invalid node type {subject}')
# predicate is valid
dom, rng = self._parse_predicate_expression(node.predicate)
# subject is a subtype of the predicate's domain
if not subject <= dom:
raise errors.ConsistencyError(f'Expected type {dom}, found {subject}')
# node.count is a numerical expression
self._parse_numerical_expression(node.count, self.schema.literal(ns.xsd.numerical))
def _equals(self, node: ast.filter.Equals, subject: bsc.types._Vertex, numerical: bool = False):
# subject is a literal
#if not isinstance(subject, bsc.Literal):
# raise errors.ConsistencyError(f'Expected a literal, found {subject}')
if isinstance(subject, bsc.Node):
# FIXME: How to handle this case?
# FIXME: How to check if a NodeType is acceptable?
# FIXME: Maybe use flags to control what is expected as node identifiers?
from bsfs.graph.nodes import Nodes # FIXME
if not isinstance(node.value, Nodes) and not isinstance(node.value, URI):
raise errors.ConsistencyError(f'Expected a Nodes or URI, found {node.value}')
elif isinstance(subject, bsc.Literal):
# literal exists in the schema
if subject not in self.schema.literals:
raise errors.ConsistencyError(f'Invalid literal type {subject}')
else:
# FIXME:
raise errors.ConsistencyError(f'Expected a literal, found {subject}')
# node.value is numeric (if requested)
if numerical and not isinstance(node.value, float) and not isinstance(node.value, int):
raise errors.ConsistencyError(f'Expected a numerical value (int or float), found {node.value}')
# NOTE: We cannot check if node.value agrees with the subject since we don't know
# all literal types, their hierarchy, and how the backend converts datatypes.
def _substring(self, node: ast.filter.Substring, subject: bsc.types._Vertex):
# subject is a literal
if not isinstance(subject, bsc.Literal):
raise errors.ConsistencyError(f'Expected a literal, found {subject}')
# literal exists in the schema
if subject not in self.schema.literals:
raise errors.ConsistencyError(f'Invalid literal type {subject}')
# node.value matches literal datatype
if not subject.is_a(ns.xsd.string):
raise errors.ConsistencyError(f'Expected a string literal, found {subject}')
def _lessThan(self, node: ast.filter.LessThan, subject: bsc.types._Vertex):
# subject is a literal
if not isinstance(subject, bsc.Literal):
raise errors.ConsistencyError(f'Expected a literal, found {subject}')
# literal exists in the schema
if subject not in self.schema.literals:
raise errors.ConsistencyError(f'Invalid literal type {subject}')
# subject is numerical
if not subject.is_a(ns.xsd.numerical):
raise errors.ConsistencyError(f'Expected a numerical literal, found {subject}')
def _greaterThan(self, node: ast.filter.GreaterThan, subject: bsc.types._Vertex):
# subject is a literal
if not isinstance(subject, bsc.Literal):
raise errors.ConsistencyError(f'Expected a literal, found {subject}')
# literal exists in the schema
if subject not in self.schema.literals:
raise errors.ConsistencyError(f'Invalid literal type {subject}')
# subject is numerical
if not subject.is_a(ns.xsd.numerical):
raise errors.ConsistencyError(f'Expected a numerical literal, found {subject}')
def _predicate(self, node: ast.filter.Predicate):
try:
# predicate exists in the schema
pred = self.schema.predicate(node.predicate)
except KeyError:
raise errors.ConsistencyError(f'') # FIXME
if node.reverse:
return pred.range, pred.domain
else:
return pred.domain, pred.range
def _oneOf(self, node: ast.filter.OneOf):
dom, rng = None, None
for pred in node:
try:
# parse child expression
subdom, subrng = self._parse_predicate_expression(pred)
# domain and range must be related across all child expressions
if not subdom <= dom and not subdom >= dom:
raise errors.ConsistencyError(f'') # FIXME
if not subrng <= rng and not subrng >= rng:
raise errors.ConsistencyError(f'') # FIXME
# determine overall domain and range
if dom is None or subdom < dom: # pick most specific domain
dom = subdom
if rng is None or subrng > rng: # pick most generic range
rng = subrng
except KeyError:
raise errors.ConsistencyError(f'')
return dom, rng
## EOF ##
|
