aboutsummaryrefslogtreecommitdiffstats
path: root/bsfs/query/ast/fetch.py
blob: 66d94e1196e67554efcb2af60b5153f851da8b6f (plain)
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

# imports
from collections import abc
import typing

# bsfs imports
from bsfs.utils import URI, typename, normalize_args

# exports
__all__ : typing.Sequence[str] = (
    'All',
    'Fetch',
    'FetchExpression',
    'Node',
    'This',
    'Value',
    )


## code ##

class FetchExpression(abc.Hashable):
    """Generic Fetch expression."""

    def __repr__(self) -> str:
        """Return the expressions's string representation."""
        return f'{typename(self)}()'

    def __hash__(self) -> int:
        """Return the expression's integer representation."""
        return hash(type(self))

    def __eq__(self, other: typing.Any) -> bool:
        """Return True if *self* and *other* are equivalent."""
        return isinstance(other, type(self))


class All(FetchExpression):
    """Fetch all child expressions."""

    # child expressions.
    expr: typing.Set[FetchExpression]

    def __init__(self, *expr):
        # unpack child expressions
        unfolded = set(normalize_args(*expr))
        # check child expressions
        if len(unfolded) == 0:
            raise AttributeError('expected at least one expression, found none')
        if not all(isinstance(itm, FetchExpression) for itm in unfolded):
            raise TypeError(expr)
        # initialize
        super().__init__()
        # assign members
        self.expr = unfolded

    def __iter__(self) -> typing.Iterator[FetchExpression]:
        return iter(self.expr)

    def __len__(self) -> int:
        return len(self.expr)

    def __repr__(self) -> str:
        return f'{typename(self)}({self.expr})'

    def __hash__(self) -> int:
        return hash((super().__hash__(), tuple(sorted(self.expr, key=repr))))

    def __eq__(self, other: typing.Any) -> bool:
        return super().__eq__(other) and self.expr == other.expr


class _Branch(FetchExpression):
    """Branch along a predicate."""

    # FIXME: Use a Predicate (like in ast.filter) so that we can also reverse them!

    # predicate to follow.
    predicate: URI

    def __init__(self, predicate: URI):
        if not isinstance(predicate, URI):
            raise TypeError(predicate)
        self.predicate = predicate

    def __repr__(self) -> str:
        return f'{typename(self)}({self.predicate})'

    def __hash__(self) -> int:
        return hash((super().__hash__(), self.predicate))

    def __eq__(self, other: typing.Any) -> bool:
        return super().__eq__(other) and self.predicate == other.predicate


class Fetch(_Branch):
    """Follow a predicate before evaluating a child epxression."""

    # child expression.
    expr: FetchExpression

    def __init__(self, predicate: URI, expr: FetchExpression):
        # check child expressions
        if not isinstance(expr, FetchExpression):
            raise TypeError(expr)
        # initialize
        super().__init__(predicate)
        # assign members
        self.expr = expr

    def __repr__(self) -> str:
        return f'{typename(self)}({self.predicate}, {self.expr})'

    def __hash__(self) -> int:
        return hash((super().__hash__(), self.expr))

    def __eq__(self, other: typing.Any) -> bool:
        return super().__eq__(other) and self.expr == other.expr


class _Named(_Branch):
    """Fetch a (named) symbol at a predicate."""

    # symbol name.
    name: str

    def __init__(self, predicate: URI, name: str):
        super().__init__(predicate)
        self.name = str(name)

    def __repr__(self) -> str:
        return f'{typename(self)}({self.predicate}, {self.name})'

    def __hash__(self) -> int:
        return hash((super().__hash__(), self.name))

    def __eq__(self, other: typing.Any) -> bool:
        return super().__eq__(other) and self.name == other.name


class Node(_Named): # pylint: disable=too-few-public-methods
    """Fetch a Node at a predicate."""
    # FIXME: Is this actually needed?


class Value(_Named): # pylint: disable=too-few-public-methods
    """Fetch a Literal at a predicate."""


class This(FetchExpression):
    """Fetch the current Node."""

    # symbol name.
    name: str

    def __init__(self, name: str):
        super().__init__()
        self.name = str(name)

    def __repr__(self) -> str:
        return f'{typename(self)}({self.name})'

    def __hash__(self) -> int:
        return hash((super().__hash__(), self.name))

    def __eq__(self, other: typing.Any) -> bool:
        return super().__eq__(other) and self.name == other.name

## EOF ##