​
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.