Skip to content

graphcast.filter.onto

Filter expression system for database queries.

This module provides a flexible system for creating and evaluating filter expressions that can be translated into different database query languages (AQL, Cypher, Python). It includes classes for logical operators, comparison operators, and filter clauses.

Key Components
  • LogicalOperator: Enum for logical operations (AND, OR, NOT, IMPLICATION)
  • ComparisonOperator: Enum for comparison operations (==, !=, >, <, etc.)
  • AbsClause: Abstract base class for filter clauses
  • LeafClause: Concrete clause for field comparisons
  • Clause: Composite clause combining multiple sub-clauses
  • Expression: Factory class for creating filter expressions from dictionaries
Example

expr = Expression.from_dict({ ... "AND": [ ... {"field": "age", "cmp_operator": ">=", "value": 18}, ... {"field": "status", "cmp_operator": "==", "value": "active"} ... ] ... })

Converts to: "age >= 18 AND status == 'active'"

AbsClause dataclass

Bases: BaseDataclass

Abstract base class for filter clauses.

This class defines the interface for all filter clauses, requiring implementation of the call method to evaluate or render the clause.

Source code in graphcast/filter/onto.py 
 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
@dataclasses.dataclass
class AbsClause(BaseDataclass, metaclass=ABCMeta):
    """Abstract base class for filter clauses.

    This class defines the interface for all filter clauses, requiring
    implementation of the __call__ method to evaluate or render the clause.
    """

    @abstractmethod
    def __call__(
        self,
        doc_name,
        kind: ExpressionFlavor = ExpressionFlavor.ARANGO,
        **kwargs,
    ):
        """Evaluate or render the clause.

        Args:
            doc_name: Name of the document variable in the query
            kind: Target expression flavor (ARANGO, NEO4J, PYTHON)
            **kwargs: Additional arguments for evaluation

        Returns:
            str: Rendered clause in the target language
        """
        pass

__call__(doc_name, kind=ExpressionFlavor.ARANGO, **kwargs) abstractmethod

Evaluate or render the clause.

Parameters:

Name Type Description Default
doc_name

Name of the document variable in the query

 required
kind ExpressionFlavor

Target expression flavor (ARANGO, NEO4J, PYTHON)

 ARANGO
**kwargs

Additional arguments for evaluation

 {}

Returns:

Name Type Description
str

Rendered clause in the target language
Source code in graphcast/filter/onto.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@abstractmethod
def __call__(
    self,
    doc_name,
    kind: ExpressionFlavor = ExpressionFlavor.ARANGO,
    **kwargs,
):
    """Evaluate or render the clause.

    Args:
        doc_name: Name of the document variable in the query
        kind: Target expression flavor (ARANGO, NEO4J, PYTHON)
        **kwargs: Additional arguments for evaluation

    Returns:
        str: Rendered clause in the target language
    """
    pass

Clause dataclass

Bases: AbsClause

Composite clause combining multiple sub-clauses.

This class represents a logical combination of multiple filter clauses, such as "clause1 AND clause2" or "NOT clause1".

Attributes:

Name Type Description
operator LogicalOperator

Logical operator to combine clauses
deps list[AbsClause]

List of dependent clauses
Source code in graphcast/filter/onto.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
@dataclasses.dataclass
class Clause(AbsClause):
    """Composite clause combining multiple sub-clauses.

    This class represents a logical combination of multiple filter clauses,
    such as "clause1 AND clause2" or "NOT clause1".

    Attributes:
        operator: Logical operator to combine clauses
        deps: List of dependent clauses
    """

    operator: LogicalOperator
    deps: list[AbsClause]

    def __call__(
        self,
        doc_name="doc",
        kind: ExpressionFlavor = ExpressionFlavor.ARANGO,
        **kwargs,
    ):
        """Render the composite clause in the target language.

        Args:
            doc_name: Document variable name
            kind: Target expression flavor
            **kwargs: Additional arguments

        Returns:
            str: Rendered clause

        Raises:
            ValueError: If operator and dependencies don't match
        """
        if kind == ExpressionFlavor.ARANGO or kind == ExpressionFlavor.ARANGO:
            return self._cast_generic(doc_name=doc_name, kind=kind)
        elif kind == ExpressionFlavor.PYTHON:
            return self._cast_python(kind=kind, **kwargs)

    def _cast_generic(self, doc_name, kind):
        """Render the clause in a generic format.

        Args:
            doc_name: Document variable name
            kind: Target expression flavor

        Returns:
            str: Rendered clause

        Raises:
            ValueError: If operator and dependencies don't match
        """
        if len(self.deps) == 1:
            if self.operator == LogicalOperator.NOT:
                return f"{self.operator} {self.deps[0](kind=kind, doc_name=doc_name)}"
            else:
                raise ValueError(
                    f" length of deps = {len(self.deps)} but operator is not"
                    f" {LogicalOperator.NOT}"
                )
        else:
            return f" {self.operator} ".join(
                [item(kind=kind, doc_name=doc_name) for item in self.deps]
            )

    def _cast_python(self, kind, **kwargs):
        """Evaluate the clause in Python.

        Args:
            kind: Expression flavor
            **kwargs: Additional arguments

        Returns:
            bool: Evaluation result

        Raises:
            ValueError: If operator and dependencies don't match
        """
        if len(self.deps) == 1:
            if self.operator == LogicalOperator.NOT:
                return not self.deps[0](kind=kind, **kwargs)
            else:
                raise ValueError(
                    f" length of deps = {len(self.deps)} but operator is not"
                    f" {LogicalOperator.NOT}"
                )
        else:
            return OperatorMapping[self.operator](
                [item(kind=kind, **kwargs) for item in self.deps]
            )

__call__(doc_name='doc', kind=ExpressionFlavor.ARANGO, **kwargs)

Render the composite clause in the target language.

Parameters:

Name Type Description Default
doc_name

Document variable name

 'doc'
kind ExpressionFlavor

Target expression flavor

 ARANGO
**kwargs

Additional arguments

 {}

Returns:

Name Type Description
str

Rendered clause

Raises:

Type Description
ValueError

If operator and dependencies don't match
Source code in graphcast/filter/onto.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def __call__(
    self,
    doc_name="doc",
    kind: ExpressionFlavor = ExpressionFlavor.ARANGO,
    **kwargs,
):
    """Render the composite clause in the target language.

    Args:
        doc_name: Document variable name
        kind: Target expression flavor
        **kwargs: Additional arguments

    Returns:
        str: Rendered clause

    Raises:
        ValueError: If operator and dependencies don't match
    """
    if kind == ExpressionFlavor.ARANGO or kind == ExpressionFlavor.ARANGO:
        return self._cast_generic(doc_name=doc_name, kind=kind)
    elif kind == ExpressionFlavor.PYTHON:
        return self._cast_python(kind=kind, **kwargs)

ComparisonOperator

Bases: BaseEnum

Comparison operators for field comparisons.

Attributes:

Name Type Description
NEQ

Not equal (!=)
EQ

Equal (==)
GE

Greater than or equal (>=)
LE

Less than or equal (<=)
GT

Greater than (>)
LT

Less than (<)
IN

Membership test (IN)
Source code in graphcast/filter/onto.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class ComparisonOperator(BaseEnum):
    """Comparison operators for field comparisons.

    Attributes:
        NEQ: Not equal (!=)
        EQ: Equal (==)
        GE: Greater than or equal (>=)
        LE: Less than or equal (<=)
        GT: Greater than (>)
        LT: Less than (<)
        IN: Membership test (IN)
    """

    NEQ = "!="
    EQ = "=="
    GE = ">="
    LE = "<="
    GT = ">"
    LT = "<"
    IN = "IN"

Expression dataclass

Bases: AbsClause

Factory class for creating filter expressions.

This class provides methods to create filter expressions from dictionaries and evaluate them in different languages.

Source code in graphcast/filter/onto.py 
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
@dataclasses.dataclass
class Expression(AbsClause):
    """Factory class for creating filter expressions.

    This class provides methods to create filter expressions from dictionaries
    and evaluate them in different languages.
    """

    @classmethod
    def from_dict(cls, current):
        """Create a filter expression from a dictionary.

        Args:
            current: Dictionary or list representing the filter expression

        Returns:
            AbsClause: Created filter expression

        Example:
            >>> expr = Expression.from_dict({
            ...     "AND": [
            ...         {"field": "age", "cmp_operator": ">=", "value": 18},
            ...         {"field": "status", "cmp_operator": "==", "value": "active"}
            ...     ]
            ... })
        """
        if isinstance(current, list):
            if current[0] in ComparisonOperator:
                return LeafClause(*current)
            elif current[0] in LogicalOperator:
                return Clause(*current)
        elif isinstance(current, dict):
            k = list(current.keys())[0]
            if k in LogicalOperator:
                clauses = [cls.from_dict(v) for v in current[k]]
                return Clause(operator=k, deps=clauses)
            else:
                return LeafClause(**current)

    def __call__(
        self,
        doc_name="doc",
        kind: ExpressionFlavor = ExpressionFlavor.ARANGO,
        **kwargs,
    ):
        """Evaluate the expression in the target language.

        Args:
            doc_name: Document variable name
            kind: Target expression flavor
            **kwargs: Additional arguments

        Returns:
            str: Rendered expression
        """
        pass

__call__(doc_name='doc', kind=ExpressionFlavor.ARANGO, **kwargs)

Evaluate the expression in the target language.

Parameters:

Name Type Description Default
doc_name

Document variable name

 'doc'
kind ExpressionFlavor

Target expression flavor

 ARANGO
**kwargs

Additional arguments

 {}

Returns:

Name Type Description
str

Rendered expression
Source code in graphcast/filter/onto.py 
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def __call__(
    self,
    doc_name="doc",
    kind: ExpressionFlavor = ExpressionFlavor.ARANGO,
    **kwargs,
):
    """Evaluate the expression in the target language.

    Args:
        doc_name: Document variable name
        kind: Target expression flavor
        **kwargs: Additional arguments

    Returns:
        str: Rendered expression
    """
    pass

from_dict(current) classmethod

Create a filter expression from a dictionary.

Parameters:

Name Type Description Default
current

Dictionary or list representing the filter expression

 required

Returns:

Name Type Description
AbsClause

Created filter expression
Example

expr = Expression.from_dict({ ... "AND": [ ... {"field": "age", "cmp_operator": ">=", "value": 18}, ... {"field": "status", "cmp_operator": "==", "value": "active"} ... ] ... })

Source code in graphcast/filter/onto.py 
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
@classmethod
def from_dict(cls, current):
    """Create a filter expression from a dictionary.

    Args:
        current: Dictionary or list representing the filter expression

    Returns:
        AbsClause: Created filter expression

    Example:
        >>> expr = Expression.from_dict({
        ...     "AND": [
        ...         {"field": "age", "cmp_operator": ">=", "value": 18},
        ...         {"field": "status", "cmp_operator": "==", "value": "active"}
        ...     ]
        ... })
    """
    if isinstance(current, list):
        if current[0] in ComparisonOperator:
            return LeafClause(*current)
        elif current[0] in LogicalOperator:
            return Clause(*current)
    elif isinstance(current, dict):
        k = list(current.keys())[0]
        if k in LogicalOperator:
            clauses = [cls.from_dict(v) for v in current[k]]
            return Clause(operator=k, deps=clauses)
        else:
            return LeafClause(**current)

LeafClause dataclass

Bases: AbsClause

Concrete clause for field comparisons.

This class represents a single field comparison operation, such as "field >= value" or "field IN [values]".

Attributes:

Name Type Description
cmp_operator ComparisonOperator | None

Comparison operator to use
value list

Value(s) to compare against
field str | None

Field name to compare
operator str | None

Optional operator to apply before comparison
Source code in graphcast/filter/onto.py 
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
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
@dataclasses.dataclass
class LeafClause(AbsClause):
    """Concrete clause for field comparisons.

    This class represents a single field comparison operation, such as
    "field >= value" or "field IN [values]".

    Attributes:
        cmp_operator: Comparison operator to use
        value: Value(s) to compare against
        field: Field name to compare
        operator: Optional operator to apply before comparison
    """

    cmp_operator: ComparisonOperator | None = None
    value: list = dataclasses.field(default_factory=list)
    field: str | None = None
    operator: str | None = None

    def __post_init__(self):
        """Convert single value to list if necessary."""
        if not isinstance(self.value, list):
            self.value = [self.value]

    def __call__(
        self,
        doc_name="doc",
        kind: ExpressionFlavor = ExpressionFlavor.ARANGO,
        **kwargs,
    ):
        """Render the leaf clause in the target language.

        Args:
            doc_name: Name of the document variable
            kind: Target expression flavor
            **kwargs: Additional arguments

        Returns:
            str: Rendered clause

        Raises:
            ValueError: If kind is not implemented
        """
        if not self.value:
            logger.warning(f"for {self} value is not set : {self.value}")
        if kind == ExpressionFlavor.ARANGO:
            assert self.cmp_operator is not None
            return self._cast_arango(doc_name)
        elif kind == ExpressionFlavor.NEO4J:
            assert self.cmp_operator is not None
            return self._cast_cypher(doc_name)
        elif kind == ExpressionFlavor.PYTHON:
            return self._cast_python(**kwargs)
        else:
            raise ValueError(f"kind {kind} not implemented")

    def _cast_value(self):
        """Format the comparison value for query rendering.

        Returns:
            str: Formatted value string
        """
        value = f"{self.value[0]}" if len(self.value) == 1 else f"{self.value}"
        if len(self.value) == 1:
            if isinstance(self.value[0], str):
                value = f'"{self.value[0]}"'
            elif self.value[0] is None:
                value = "null"
            else:
                value = f"{self.value[0]}"
        return value

    def _cast_arango(self, doc_name):
        """Render the clause in AQL format.

        Args:
            doc_name: Document variable name

        Returns:
            str: AQL clause
        """
        const = self._cast_value()

        lemma = f"{self.cmp_operator} {const}"
        if self.operator is not None:
            lemma = f"{self.operator} {lemma}"

        if self.field is not None:
            lemma = f'{doc_name}["{self.field}"] {lemma}'
        return lemma

    def _cast_cypher(self, doc_name):
        """Render the clause in Cypher format.

        Args:
            doc_name: Document variable name

        Returns:
            str: Cypher clause
        """
        const = self._cast_value()
        if self.cmp_operator == ComparisonOperator.EQ:
            cmp_operator = "="
        else:
            cmp_operator = self.cmp_operator
        lemma = f"{cmp_operator} {const}"
        if self.operator is not None:
            lemma = f"{self.operator} {lemma}"

        if self.field is not None:
            lemma = f"{doc_name}.{self.field} {lemma}"
        return lemma

    def _cast_python(self, **kwargs):
        """Evaluate the clause in Python.

        Args:
            **kwargs: Additional arguments

        Returns:
            bool: Evaluation result
        """
        field = kwargs.pop(self.field, None)
        if field is not None:
            foo = getattr(field, self.operator)
            return foo(self.value[0])
        else:
            return False

__call__(doc_name='doc', kind=ExpressionFlavor.ARANGO, **kwargs)

Render the leaf clause in the target language.

Parameters:

Name Type Description Default
doc_name

Name of the document variable

 'doc'
kind ExpressionFlavor

Target expression flavor

 ARANGO
**kwargs

Additional arguments

 {}

Returns:

Name Type Description
str

Rendered clause

Raises:

Type Description
ValueError

If kind is not implemented
Source code in graphcast/filter/onto.py 
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
def __call__(
    self,
    doc_name="doc",
    kind: ExpressionFlavor = ExpressionFlavor.ARANGO,
    **kwargs,
):
    """Render the leaf clause in the target language.

    Args:
        doc_name: Name of the document variable
        kind: Target expression flavor
        **kwargs: Additional arguments

    Returns:
        str: Rendered clause

    Raises:
        ValueError: If kind is not implemented
    """
    if not self.value:
        logger.warning(f"for {self} value is not set : {self.value}")
    if kind == ExpressionFlavor.ARANGO:
        assert self.cmp_operator is not None
        return self._cast_arango(doc_name)
    elif kind == ExpressionFlavor.NEO4J:
        assert self.cmp_operator is not None
        return self._cast_cypher(doc_name)
    elif kind == ExpressionFlavor.PYTHON:
        return self._cast_python(**kwargs)
    else:
        raise ValueError(f"kind {kind} not implemented")

__post_init__()

Convert single value to list if necessary.

Source code in graphcast/filter/onto.py 
142
143
144
145
def __post_init__(self):
    """Convert single value to list if necessary."""
    if not isinstance(self.value, list):
        self.value = [self.value]

LogicalOperator

Bases: BaseEnum

Logical operators for combining filter conditions.

Attributes:

Name Type Description
AND

Logical AND operation
OR

Logical OR operation
NOT

Logical NOT operation
IMPLICATION

Logical IF-THEN operation
Source code in graphcast/filter/onto.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class LogicalOperator(BaseEnum):
    """Logical operators for combining filter conditions.

    Attributes:
        AND: Logical AND operation
        OR: Logical OR operation
        NOT: Logical NOT operation
        IMPLICATION: Logical IF-THEN operation
    """

    AND = "AND"
    OR = "OR"
    NOT = "NOT"
    IMPLICATION = "IF_THEN"

implication(ops)

Evaluate logical implication (IF-THEN).

Parameters:

Name Type Description Default
ops

Tuple of (antecedent, consequent)

 required

Returns:

Name Type Description
bool

True if antecedent is False or consequent is True
Source code in graphcast/filter/onto.py 
51
52
53
54
55
56
57
58
59
60
61
def implication(ops):
    """Evaluate logical implication (IF-THEN).

    Args:
        ops: Tuple of (antecedent, consequent)

    Returns:
        bool: True if antecedent is False or consequent is True
    """
    a, b = ops
    return b if a else True