Graph data store

Graph data store interface module.

This module defines the BaseGraphDataStore abstract base class, which provides an interface for graph database operations.

Authors

Berty C L Tobing (berty.c.l.tobing@gdplabs.id)

References

NONE

BaseGraphDataStore

Bases: ABC

Abstract base class for an async graph data store interface.

This class defines the asynchronous interface for all graph data store implementations. It provides methods for creating, updating, and querying graph data.

close() abstractmethod async

Close the graph data store.

delete_node(label, identifier_key, identifier_value) abstractmethod async

Delete a node from the graph.

Parameters:

Name	Type	Description	Default
label	str	The label of the node.	required
identifier_key	str	The key of the identifier.	required
identifier_value	str	The identifier of the node.	required

Returns:

Name	Type	Description
Any	Any	The result of the operation.

delete_relationship(node_source_key, node_source_value, relation, node_target_key, node_target_value) abstractmethod async

Delete a relationship between two nodes in the graph.

Parameters:

Name	Type	Description	Default
node_source_key	str	The key of the source node.	required
node_source_value	str	The identifier of the source node.	required
relation	str	The type of the relationship.	required
node_target_key	str	The key of the target node.	required
node_target_value	str	The identifier of the target node.	required

Returns:

Name	Type	Description
Any	Any	The result of the operation.

query(query, parameters=None) abstractmethod async

Query the graph data store.

Parameters:

Name	Type	Description	Default
query	str	The query to be executed.	required
parameters	dict[str, Any] | None	The parameters of the query. Defaults to None.	None

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: The result of the query as a list of dictionaries.

traverse_graph(node_properties, extracted_node_properties=None, extracted_relationship_properties=None, depth=3) abstractmethod async

Traverse graph from a node with specified properties, ignoring relationship's direction, up to a given depth.

Example

nodes, relationships = await graph_data_store.traverse_graph(
    node_properties={"name": "John Doe"},
    extracted_node_properties=["name", "age"],
    extracted_relationship_properties=["since"],
    depth=1
)

Means starting from the node with property name equal to "John Doe", traverse the graph up to depth 1, extracting the name and age properties from nodes and the since property from relationships.

nodes, relationships = await graph_data_store.traverse_graph(
    node_properties={"name": "John Doe"},
    depth=2
)

Means starting from the node with property name equal to "John Doe", traverse the graph up to depth 2, extracting all properties from nodes and relationships.

Parameters:

Name	Type	Description	Default
node_properties	dict[str, Any]	The properties of the starting node.	required
extracted_node_properties	list[str] | None	The properties to extract from nodes during traversal. If None or empty list, all node properties will be returned. Defaults to None.	None
extracted_relationship_properties	list[str] | None	The properties to extract from relationships during traversal. If None or empty list, all relationship properties will be returned. Defaults to None.	None
depth	int	The depth of traversal. Defaults to 3.	3

Returns:

Type	Description
list[dict[str, Any]]	tuple[list[dict[str, Any]], list[dict[str, Any]]]: A tuple containing two lists: - List of nodes with their extracted properties. - List of relationships with their extracted properties.
list[dict[str, Any]]	Example return value:
tuple[list[dict[str, Any]], list[dict[str, Any]]]	nodes = [ { "id": 1001, "labels": ["Person"], "properties": { "name": "John Doe", "age": 30, "occupation": "Engineer" } }, { "id": 2001, "labels": ["Company"], "properties": { "name": "TechCorp", "industry": "Technology", "employees": 500 } }
tuple[list[dict[str, Any]], list[dict[str, Any]]]	]
tuple[list[dict[str, Any]], list[dict[str, Any]]]	relationships = [ { "id": 5002, "type": "FRIEND_OF", "start_node": 1001, "end_node": 1002, "properties": { "since": "2018-05-20", "closeness": 8 } }
tuple[list[dict[str, Any]], list[dict[str, Any]]]	]

upsert_node(label, identifier_key, identifier_value, properties) abstractmethod async

Upsert a node in the graph.

Parameters:

Name	Type	Description	Default
label	str	The label of the node.	required
identifier_key	str	The key of the identifier.	required
identifier_value	str	The value of the identifier.	required
properties	dict[str, Any] | None	The properties of the node. Defaults to None.	required

Returns:

Name	Type	Description
Any	Any	The result of the operation.

upsert_relationship(node_source_key, node_source_value, relation, node_target_key, node_target_value, properties) abstractmethod async

Upsert a relationship between two nodes in the graph.

Parameters:

Name	Type	Description	Default
node_source_key	str	The key of the source node.	required
node_source_value	str	The value of the source node.	required
relation	str	The type of the relationship.	required
node_target_key	str	The key of the target node.	required
node_target_value	str	The value of the target node.	required
properties	dict[str, Any] | None	The properties of the relationship. Defaults to None.	required

Returns:

Name	Type	Description
Any	Any	The result of the operation.