""" Part of the BlackStar filesystem (bsfs) module. A copy of the license is provided with the project. Author: Matthias Baumgartner, 2022 """ # 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 ##