​
NebulaGraph

class NebulaGraph(BaseGraphStorage):

​
init

def __init__(
    self,
    host,
    username,
    password,
    space,
    port = 9669,
    timeout = 10000
):

Initializes the NebulaGraph client.

Parameters:

  • host (str): The host address of the NebulaGraph service.
  • username (str): The username for authentication.
  • password (str): The password for authentication.
  • space (str): The graph space to use. If it doesn’t exist, a new one will be created.
  • port (int, optional): The port number for the connection. (default: :obj:9669)
  • timeout (int, optional): The connection timeout in milliseconds. (default: :obj:10000)

​
_init_connection_pool

def _init_connection_pool(self):

Returns:

ConnectionPool: A connection pool instance.

​
_get_session

def _get_session(self):

Returns:

Session: A session object connected to NebulaGraph.

​
get_client

def get_client(self):

Get the underlying graph storage client.

​
query

def query(self, query: str):

Execute a query on the graph store.

Parameters:

  • query (str): The Cypher-like query to be executed.

Returns:

ResultSet: The result set of the query execution.

​
get_relationship_types

def get_relationship_types(self):

Returns:

List[str]: A list of relationship (edge) type names.

​
add_graph_elements

def add_graph_elements(self, graph_elements: List[GraphElement]):

Add graph elements (nodes and relationships) to the graph.

Parameters:

  • graph_elements (List[GraphElement]): A list of graph elements containing nodes and relationships.

​
ensure_edge_type_exists

def ensure_edge_type_exists(self, edge_type: str, time_label: Optional[str] = None):

Ensures that a specified edge type exists in the NebulaGraph database. If the edge type already exists, this method does nothing.

Parameters:

  • edge_type (str): The name of the edge type to be created.
  • time_label (str, optional): A specific timestamp to set as the default value for the time label property. If not provided, no timestamp will be added. (default: :obj:None)

​
ensure_tag_exists

def ensure_tag_exists(self, tag_name: str, time_label: Optional[str] = None):

Ensures a tag is created in the NebulaGraph database. If the tag already exists, it does nothing.

Parameters:

  • tag_name (str): The name of the tag to be created.
  • time_label (str, optional): A specific timestamp to set as the default value for the time label property. If not provided, no timestamp will be added. (default: :obj:None)

​
add_node

def add_node(
    self,
    node_id: str,
    tag_name: str,
    time_label: Optional[str] = None
):

Add a node with the specified tag and properties.

Parameters:

  • node_id (str): The ID of the node.
  • tag_name (str): The tag name of the node.
  • time_label (str, optional): A specific timestamp to set for the node’s time label property. If not provided, no timestamp will be added. (default: :obj:None)

​
_extract_nodes

def _extract_nodes(self, graph_elements: List[Any]):

Extracts unique nodes from graph elements.

Parameters:

  • graph_elements (List[Any]): A list of graph elements containing nodes.

Returns:

List[Dict]: A list of dictionaries representing nodes.

​
_extract_relationships

def _extract_relationships(self, graph_elements: List[Any]):

Extracts relationships from graph elements.

Parameters:

  • graph_elements (List[Any]): A list of graph elements containing relationships.

Returns:

List[Dict]: A list of dictionaries representing relationships.

​
refresh_schema

def refresh_schema(self):

Refreshes the schema by fetching the latest schema details.

​
get_structured_schema

def get_structured_schema(self):

Returns:

Dict[str, Any]: A dictionary representing the structured schema.

​
get_schema

def get_schema(self):

Returns:

str: A string describing the schema.

​
get_indexes

def get_indexes(self):

Returns:

List[str]: A list of tag index names.

​
add_triplet

def add_triplet(
    self,
    subj: str,
    obj: str,
    rel: str,
    time_label: Optional[str] = None
):

Adds a relationship (triplet) between two entities in the Nebula Graph database.

Parameters:

  • subj (str): The identifier for the subject entity.
  • obj (str): The identifier for the object entity.
  • rel (str): The relationship between the subject and object.
  • time_label (str, optional): A specific timestamp to set for the time label property of the relationship. If not provided, no timestamp will be added. (default: :obj:None)

​
delete_triplet

def delete_triplet(
    self,
    subj: str,
    obj: str,
    rel: str
):

Deletes a specific triplet (relationship between two entities) from the Nebula Graph database.

Parameters:

  • subj (str): The identifier for the subject entity.
  • obj (str): The identifier for the object entity.
  • rel (str): The relationship between the subject and object.

​
delete_entity

def delete_entity(self, entity_id: str):

Deletes an entity (vertex) from the graph.

Parameters:

  • entity_id (str): The identifier of the entity to be deleted.

​
_check_edges

def _check_edges(self, entity_id: str):

Checks if an entity has any remaining edges in the graph.

Parameters:

  • entity_id (str): The identifier of the entity.

Returns:

bool: :obj:True if the entity has edges, :obj:False otherwise.

​
get_node_properties

def get_node_properties(self):

Returns:

Tuple[List[str], List[Dict[str, Any]]]: A tuple where the first element is a list of node schema properties, and the second element is a list of dictionaries representing node structures.

​
get_relationship_properties

def get_relationship_properties(self):

Returns:

Tuple[List[str], List[Dict[str, Any]]]: A tuple where the first element is a list of relationship schema properties, and the second element is a list of dictionaries representing relationship structures.

​
_validate_time_label

def _validate_time_label(self, time_label: str):

Validates the format of a time label string.

Parameters:

  • time_label (str): The time label string to validate. Should be in format ‘YYYY-MM-DDThh:mm:ss’.

Returns:

str: The validated time label.