> ## Documentation Index
> Fetch the complete documentation index at: https://docs.camel-ai.org/llms.txt
> Use this file to discover all available pages before exploring further.

# Camel.storages.graph storages.nebula graph

<a id="camel.storages.graph_storages.nebula_graph" />

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph" />

## NebulaGraph

```python theme={"system"}
class NebulaGraph(BaseGraphStorage):
```

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.__init__" />

### **init**

```python theme={"system"}
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`)

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph._init_connection_pool" />

### \_init\_connection\_pool

```python theme={"system"}
def _init_connection_pool(self):
```

**Returns:**

ConnectionPool: A connection pool instance.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph._get_session" />

### \_get\_session

```python theme={"system"}
def _get_session(self):
```

**Returns:**

Session: A session object connected to NebulaGraph.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.get_client" />

### get\_client

```python theme={"system"}
def get_client(self):
```

Get the underlying graph storage client.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.query" />

### query

```python theme={"system"}
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.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.get_relationship_types" />

### get\_relationship\_types

```python theme={"system"}
def get_relationship_types(self):
```

**Returns:**

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

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.add_graph_elements" />

### add\_graph\_elements

```python theme={"system"}
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.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.ensure_edge_type_exists" />

### ensure\_edge\_type\_exists

```python theme={"system"}
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`)

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.ensure_tag_exists" />

### ensure\_tag\_exists

```python theme={"system"}
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`)

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.add_node" />

### add\_node

```python theme={"system"}
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`)

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph._extract_nodes" />

### \_extract\_nodes

```python theme={"system"}
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.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph._extract_relationships" />

### \_extract\_relationships

```python theme={"system"}
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.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.refresh_schema" />

### refresh\_schema

```python theme={"system"}
def refresh_schema(self):
```

Refreshes the schema by fetching the latest schema details.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.get_structured_schema" />

### get\_structured\_schema

```python theme={"system"}
def get_structured_schema(self):
```

**Returns:**

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

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.get_schema" />

### get\_schema

```python theme={"system"}
def get_schema(self):
```

**Returns:**

str: A string describing the schema.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.get_indexes" />

### get\_indexes

```python theme={"system"}
def get_indexes(self):
```

**Returns:**

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

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.add_triplet" />

### add\_triplet

```python theme={"system"}
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`)

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.delete_triplet" />

### delete\_triplet

```python theme={"system"}
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.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.delete_entity" />

### delete\_entity

```python theme={"system"}
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.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph._check_edges" />

### \_check\_edges

```python theme={"system"}
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.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.get_node_properties" />

### get\_node\_properties

```python theme={"system"}
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.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph.get_relationship_properties" />

### get\_relationship\_properties

```python theme={"system"}
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.

<a id="camel.storages.graph_storages.nebula_graph.NebulaGraph._validate_time_label" />

### \_validate\_time\_label

```python theme={"system"}
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.