motleycrew.storage.kuzu_graph_store
Kùzu graph store index.
Classes
|
Kuzu graph store implementation for motleycrew. |
- class motleycrew.storage.kuzu_graph_store.MotleyKuzuGraphStore(database: Any)
Bases:
MotleyGraphStoreKuzu graph store implementation for motleycrew.
- ID_ATTR = '_id'
- JSON_CONTENT_PREFIX = 'JSON__'
- PYTHON_TO_CYPHER_TYPES_MAPPING = {<class 'bool'>: 'BOOLEAN', <class 'float'>: 'DOUBLE', <class 'int'>: 'INT64', <class 'str'>: 'STRING', typing.Optional[bool]: 'BOOLEAN', typing.Optional[float]: 'DOUBLE', typing.Optional[int]: 'INT64', typing.Optional[str]: 'STRING'}
- __init__(database: Any) None
Initialize Kuzu graph store.
- Parameters:
database – Kuzu database client
- property database_path: str
- ensure_node_table(node_class: Type[MotleyGraphNode]) str
Create a table for storing nodes of that class if such does not already exist. If it does exist, create all missing columns.
- Parameters:
node_class – Node Python class.
- Returns:
Node table name.
- ensure_relation_table(from_class: Type[MotleyGraphNode], to_class: Type[MotleyGraphNode], label: str) None
Create a table for storing relations from from_node-like nodes to to_node-like nodes, if such does not already exist.
- Parameters:
from_class – Source node Python class.
to_class – Destination node Python class.
label – Relation label.
- check_node_exists_by_class_and_id(node_class: Type[MotleyGraphNode], node_id: int) bool
Check if a node of given class with given id is present in the database.
- Parameters:
node_class – Node Python class.
node_id – Node id.
- Returns:
Whether the node exists in the database.
- check_node_exists(node: MotleyGraphNode) bool
Check if the given node is present in the database.
- Parameters:
node – Node to check.
- Returns:
Whether the node exists in the database.
- check_relation_exists(from_node: MotleyGraphNode, to_node: MotleyGraphNode, label: str | None = None) bool
Check if a relation exists between two nodes with given label.
- Parameters:
from_node – Source node.
to_node – Destination node.
label – Relation label. If None, any relation is taken into account.
- Returns:
Whether the relation exists in the database.
- get_node_by_class_and_id(node_class: Type[MotleyGraphNodeType], node_id: int) MotleyGraphNodeType | None
Retrieve the node of given class with given id if it is present in the database.
- Parameters:
node_class – Node Python class.
node_id – Node id.
- Returns:
Node object or None if it does not exist.
- insert_node(node: MotleyGraphNodeType) MotleyGraphNodeType
Insert a new node and populate its id.
If the node table or some columns do not exist, this method also creates them.
- Parameters:
node – Node to insert.
- Returns:
Inserted node.
- create_relation(from_node: MotleyGraphNode, to_node: MotleyGraphNode, label: str) None
Create a relation between existing nodes.
If the relation table does not exist, this method also creates it.
- Parameters:
from_node – Source node.
to_node – Destination node.
label – Relation label.
- upsert_triplet(from_node: MotleyGraphNode, to_node: MotleyGraphNode, label: str) None
Create a relation with a given label between nodes, if such does not already exist.
If the nodes do not already exist, create them too. This method also creates and/or updates all necessary tables.
- Parameters:
from_node – Source node.
to_node – Destination node.
label – Relation label.
- delete_node(node: MotleyGraphNode) None
Delete a given node and its relations.
- Parameters:
node – Node to delete.
- update_property(node: MotleyGraphNode, property_name: str) MotleyGraphNode
Update a graph node’s property with the corresponding value from the node object.
- Parameters:
node – Node to update.
property_name – Property name to update.
- Returns:
Updated node.
- run_cypher_query(query: str, parameters: dict | None = None, container: Type[MotleyGraphNodeType] | None = None) list[list | MotleyGraphNodeType]
Run a Cypher query and return the results.
If container class is provided, deserialize the results into objects of that class.
- Parameters:
query – Cypher query.
parameters – Query parameters.
container – Node class to deserialize the results into. If None, return raw results.
- Returns:
List of query results.
- classmethod from_persist_dir(persist_dir: str, **kwargs) MotleyKuzuGraphStore
Load from persist dir.
- Parameters:
persist_dir (str) – Persist directory.
- Returns:
Graph store.
- classmethod from_dict(config_dict: Dict[str, Any]) MotleyKuzuGraphStore
Initialize graph store from configuration dictionary.
- Parameters:
config_dict – Configuration dictionary.
- Returns:
Graph store.