289 lines
7.9 KiB
Python
289 lines
7.9 KiB
Python
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
|
|
This code is based on:
|
|
https://msdl.uantwerpen.be/git/yentl/modelverse/src/master/state/modelverse_state
|
|
"""
|
|
|
|
@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) -> None:
|
|
"""
|
|
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
|