from abc import ABC, abstractmethod from typing import Any, List, Tuple, TypeVar, Optional, Union, Generator import uuid primitive_types = (int, float, str, bool) Node = str Edge = str Element = Union[Node, Edge] class State(ABC): """ Abstract base class for MvS CRUD interface defined in: http://msdl.cs.mcgill.ca/people/yentl/files/thesis.pdf """ @staticmethod def new_id() -> str: """ Generates a new UUID """ return str(uuid.uuid4()) @staticmethod def is_valid_datavalue(value: Any) -> bool: """ Checks whether value type is supported. Args: value: value whose type needs to be checked Returns: True if value type is supported, False otherwise. """ if not isinstance(value, primitive_types): return False elif isinstance(value, int) and not (-2**63 <= value <= 2**63 - 1): return False return True def purge(self): """ Implements a garbage collection routine for implementations that don't have automatic garbage collection. """ pass # ========================================================================= # CREATE # ========================================================================= @abstractmethod def create_node(self) -> Node: """ Creates node. Returns: The created node. """ pass @abstractmethod def create_edge(self, source: Element, target: Element) -> Optional[Edge]: """ Creates edge. Source and target elements should already exist. Args: source: source element of edge target: target element of edge Returns: The created edge, None if source or target element doesn't exist. """ pass @abstractmethod def create_nodevalue(self, value: Any) -> Optional[Node]: """ Creates node containing value. Args: value: value to assign to new node Returns: The created node, None if type of value is not supported. """ pass @abstractmethod def create_dict(self, source: Element, value: Any, target: Element) -> Optional[Tuple[Edge, Edge, Node]]: """ Creates named edge between two graph elements. Args: source: source element of edge value: edge label target: target element of edge Returns: Tuple containing the created edges and label node, None is source or target don't exist or value type doesn't exist. First edge goes from source to target, second from edge to label. """ pass # ========================================================================= # READ # ========================================================================= @abstractmethod def read_root(self) -> Node: """ Reads state's root node. Returns: The state's root node. """ pass @abstractmethod def read_value(self, node: Node) -> Optional[Any]: """ Reads value of given node. Args: node: node whose value to read Returns: I node exists, value stored in node, else None. """ pass @abstractmethod def read_outgoing(self, elem: Element) -> Optional[List[Edge]]: """ Retrieves edges whose source is given element. Args: elem: source element of edges to retrieve Returns: If elem exists, list of edges whose source is elem, else None. """ pass @abstractmethod def read_incoming(self, elem: Element) -> Optional[List[Edge]]: """ Retrieves edges whose target is given element. Args: elem: target element of edges to retrieve Returns: If elem exists, list of edges whose target is elem, else None. """ pass @abstractmethod def read_edge(self, edge: Edge) -> Tuple[Optional[Node], Optional[Node]]: """ Reads source and target of given edge. Args: edge: edge whose source and target to read Returns: If edge exists, tuple containing source (first) and target (second) node, else (None, None) """ pass @abstractmethod def read_dict(self, elem: Element, value: Any) -> Optional[Element]: """ Reads element connected to given element through edge with label = value. Args: elem: source element value: edge label Returns: If elem doesn't exist or no edge is found with given label, None, else target element of edge with label = value originating from source. """ pass @abstractmethod def read_dict_keys(self, elem: Element) -> Optional[List[Any]]: """ Reads labels of outgoing edges starting in given node. Args: elem: source element Returns: If elem exists, list of (unique) edge labels, else None. """ pass @abstractmethod def read_dict_edge(self, elem: Element, value: Any) -> Optional[Edge]: """ Reads edge between two elements connected through edge with label = value. Args: elem: source element value: edge label Returns: If elem doesn't exist or no edge is found with given label, None, else edge with label = value originating from source. """ pass @abstractmethod def read_dict_node(self, elem: Element, value_node: Node) -> Optional[Element]: """ Reads element connected to given element through edge with label node = value_node. Args: elem: source element value_node: edge label node Returns: If elem exists, target element of edge with label stored in value_node originating from elem, else None. """ pass @abstractmethod def read_dict_node_edge(self, elem: Element, value_node: Node) -> Optional[Edge]: """ Reads edge connecting two elements through edge with label node = value_node. Args: elem: source element value_node: edge label node Returns: If elem exists, edge with label node = value_node, originating from source, else None. """ pass @abstractmethod def read_reverse_dict(self, elem: Element, value: Any) -> Optional[List[Element]]: """ Retrieves a list of all elements that have an outgoing edge, having label = value, towards the passed element. Args: elem: target element value: edge label Returns: If elem exists, list of elements with an outgoing edge with label = value towards elem, else None. """ pass # ========================================================================= # UPDATE # ========================================================================= """ Updates are done by performing subsequent CREATE and DELETE operations: http://msdl.cs.mcgill.ca/people/yentl/files/thesis.pdf """ # ========================================================================= # DELETE # ========================================================================= @abstractmethod def delete_node(self, node: Node) -> None: """ Deletes given node from state graph. Args: node: node to be deleted Returns: None """ pass @abstractmethod def delete_edge(self, edge: Edge) -> None: """ Deletes given edge from state graph. Args: edge: edge to be deleted Returns: None """ pass