Data Operations

The GQLDB Python driver provides methods for inserting, updating, and deleting nodes and edges in the graph.

Data Methods

insert_nodes and insert_edges are dual-shape — Python dispatches on the type of the first argument at runtime:

insert_nodes_batch_auto / insert_edges_batch_auto are alternate names for the gRPC path and continue to work (not deprecated).

Choosing a path

NOTE

Custom edge _id requires WITH EDGE_ID on the target graph. This is a server-side prerequisite — the graph must have been created with CREATE GRAPH <name> WITH EDGE_ID for user-supplied edge _ids to be honored on either path. Without it, the server auto-generates edge _ids and any value passed via EdgeData.id is ignored.

Inserting Nodes (gRPC Batch)

insert_nodes_batch_auto()

Insert multiple nodes into a graph:

Python
from gqldb import GqldbClient, GqldbConfig, NodeData

config = GqldbConfig(hosts=["localhost:9000"])

with GqldbClient(config) as client:
    client.login("admin", "password")
    client.use_graph("myGraph")

    # Create node data
    nodes = [
        NodeData(
            labels=["User"],
            properties={"name": "Alice", "age": 30, "email": "[email protected]"}
        ),
        NodeData(
            labels=["User"],
            properties={"name": "Bob", "age": 25, "email": "[email protected]"}
        ),
        NodeData(
            labels=["User", "Admin"],
            properties={"name": "Charlie", "age": 35}
        )
    ]

    # Insert nodes
    result = client.insert_nodes("myGraph", nodes)

    print(f"Success: {result.success}")
    print(f"Inserted: {result.node_count} nodes")
    print(f"Node IDs: {result.node_ids}")

NodeData Class

Python
from dataclasses import dataclass, field
from typing import List, Dict, Any

@dataclass
class NodeData:
    id: str = ""                      # Optional custom node _id (auto-generated when empty)
    labels: List[str] = field(default_factory=list)
    properties: Dict[str, Any] = field(default_factory=dict)

A non-empty id is written as _id on the inserted node (both gRPC and GQL paths).

Insert Options

Python
from gqldb.types import BulkCreateNodesOptions
from gqldb import InsertType

options = BulkCreateNodesOptions(
    mode=InsertType.OVERWRITE   # Duplicate-_id semantics: NORMAL (default), OVERWRITE, or UPSERT
)

result = client.insert_nodes("myGraph", nodes, options)

Inserting Edges

insert_edges()

Insert multiple edges into a graph:

Python
from gqldb import EdgeData

edges = [
    EdgeData(
        label="Follows",
        from_node_id="u1",
        to_node_id="u2",
        properties={"since": "2023-01-15"}
    ),
    EdgeData(
        label="Follows",
        from_node_id="u2",
        to_node_id="u3",
        properties={"since": "2023-06-20"}
    ),
    EdgeData(
        label="Knows",
        from_node_id="u1",
        to_node_id="u3",
        properties={"years": 5}
    )
]

result = client.insert_edges("myGraph", edges)

print(f"Success: {result.success}")
print(f"Inserted: {result.edge_count} edges")
print(f"Skipped: {result.skipped_count}")

EdgeData Class

Python
from dataclasses import dataclass, field
from typing import Dict, Any

@dataclass
class EdgeData:
    id: str = ""                   # Optional custom edge _id (requires WITH EDGE_ID graph)
    label: str = ""                # Edge label (type)
    from_node_id: str = ""         # Source node ID
    to_node_id: str = ""           # Target node ID
    properties: Dict[str, Any] = field(default_factory=dict)

A non-empty id is written as _id on the inserted edge (both gRPC and GQL paths). The target graph must have been created with WITH EDGE_ID for the server to honor user-supplied edge _ids.

Edge Insert Options

Python
from gqldb.types import BulkCreateEdgesOptions

options = BulkCreateEdgesOptions(
    skip_invalid_nodes=True   # Skip edges with invalid endpoints
)

result = client.insert_edges("myGraph", edges, options)

GQL-based Insert (Convenience)

insert_nodes() / insert_edges()

These convenience methods generate and execute GQL INSERT statements. They don't require a bulk import session and use the session's current graph:

Python
client.use_graph("myGraph")

nodes = [
    NodeData(labels=["Person"], properties={"name": "Alice", "age": 30}),
    NodeData(labels=["Person"], properties={"name": "Bob", "age": 25}),
    # Custom _id via the id field
    NodeData(id="p3", labels=["Person"], properties={"name": "Charlie"}),
]
client.insert_nodes(nodes)

edges = [
    EdgeData(label="Knows", from_node_id="id1", to_node_id="id2", properties={"since": 2024}),
    # Custom _id (requires graph created WITH EDGE_ID)
    EdgeData(id="tx-001", label="Knows", from_node_id="id1", to_node_id="id3", properties={"since": 2025}),
]
client.insert_edges(edges)

NOTE

GQL INSERT only supports a single label per node; if NodeData.labels has multiple entries, only the first is used in the GQL path. Use the gRPC path for multi-label nodes.

Per-call Configuration (InsertConfig)

The GQL-path insert_nodes(nodes, …) / insert_edges(edges, …) accept an optional InsertConfig for per-call graph routing and insert mode, without changing session state:

Python
from gqldb import InsertConfig, InsertType

# Target a specific graph without use_graph()
cfg = InsertConfig(
    graph_name="myGraph",
    insert_type=InsertType.OVERWRITE,   # NORMAL (default), OVERWRITE, or UPSERT
    timeout=60,                         # optional per-call timeout (seconds)
)
client.insert_nodes(nodes, cfg)
client.insert_edges(edges, cfg)

InsertType semantics

OVERWRITE and UPSERT are different semantics on existing rows; they are not interchangeable.

All other convenience methods accept QueryConfig the same way:

Python
from gqldb import QueryConfig

client.create_node_label("User", props, config=QueryConfig(graph_name="graphA"))
client.show_node_labels(config=QueryConfig(graph_name="graphB"))
client.gql("MATCH (n) RETURN n", config=QueryConfig(graph_name="graphC", timeout=10))

Passing a per-call config is thread-safe: multiple threads can target different graphs via their own config objects without interfering.

Deleting Nodes

Nodes are deleted either by _id list (delete_nodes_by_ids) or by labels and/or a WHERE condition (delete_nodes_by_condition). Both emit DETACH DELETE (attached edges are removed too), return a Response, and report the count via response.rows_affected. They operate on the session's current graph; pass a DeleteConfig(graph_name=...) to target a different graph without use_graph().

delete_nodes_by_ids()

Python
client.use_graph("myGraph")

# Delete by _id list
response = client.delete_nodes_by_ids(["u1", "u2", "u3"])
print(f"Deleted: {response.rows_affected} nodes")

delete_nodes_by_condition()

Python
# Delete by labels
response = client.delete_nodes_by_condition(labels=["TempUser"])

# Delete with a WHERE clause (omit the WHERE keyword)
response = client.delete_nodes_by_condition(labels=["User"], where="n.age < 18")
print(f"Deleted: {response.rows_affected} nodes")

# Cap the number deleted with limit
response = client.delete_nodes_by_condition(labels=["User"], where="n.active = false", limit=100)

NOTE

delete_nodes_by_condition raises ValueError if both labels and where are empty. To intentionally delete every node, pass config=DeleteConfig(allow_delete_all=True).

Deleting Edges

Edges are deleted either by _id list (delete_edges_by_ids) or by label and/or a WHERE condition (delete_edges_by_condition). Both return a Response; use response.rows_affected for the count.

delete_edges_by_ids()

Python
# Delete by _id list (requires the graph created WITH EDGE_ID / EDGE_ID ENABLED)
response = client.delete_edges_by_ids(["e1", "e2"])
print(f"Deleted: {response.rows_affected} edges")

delete_edges_by_condition()

Python
# Delete by label
response = client.delete_edges_by_condition(label="TempConnection")

# Delete with a WHERE clause (omit the WHERE keyword)
response = client.delete_edges_by_condition(label="Follows", where="e.since < '2020-01-01'")
print(f"Deleted: {response.rows_affected} edges")

NOTE

delete_edges_by_condition raises ValueError if both label and where are empty. To intentionally delete every edge, pass config=DeleteConfig(allow_delete_all=True).

Targeting a graph with DeleteConfig

DeleteConfig extends QueryConfig, so per-call graph_name works the same as InsertConfig. Set return_deleted=False on large bulk deletes to skip returning the deleted data (rows_affected is still populated):

Python
from gqldb import DeleteConfig

cfg = DeleteConfig(graph_name="myGraph", return_deleted=False)
response = client.delete_nodes_by_condition(labels=["TempUser"], config=cfg)
print(f"Deleted: {response.rows_affected} nodes")

Using GQL for Data Operations

You can also use GQL queries for data operations:

Python
# Insert with GQL
client.gql("""
    INSERT (a:User {_id: 'u1', name: 'Alice'}),
           (b:User {_id: 'u2', name: 'Bob'}),
           (a)-[:Follows {since: '2024-01-01'}]->(b)
""")

# Update with GQL
client.gql("MATCH (u:User {_id: 'u1'}) SET u.age = 31")

# Delete with GQL
client.gql("MATCH (u:User {_id: 'u1'}) DELETE u")

Result Classes

InsertNodesResult

Python
@dataclass
class InsertNodesResult:
    success: bool
    node_ids: List[str]
    node_count: int
    message: str

InsertEdgesResult

Python
@dataclass
class InsertEdgesResult:
    success: bool
    edge_ids: List[str]
    edge_count: int
    message: str
    skipped_count: int

The delete methods do not return a dedicated result class — they return a Response. Read the number of affected entities from response.rows_affected.

Error Handling

Python
from gqldb.errors import (
    GqldbError,
    InsertFailedError,
    DeleteFailedError,
    GraphNotFoundError
)

try:
    result = client.insert_nodes("myGraph", nodes)
    if not result.success:
        print(f"Insert warning: {result.message}")

except InsertFailedError as e:
    print(f"Insert failed: {e}")

except DeleteFailedError as e:
    print(f"Delete failed: {e}")

except GraphNotFoundError:
    print("Graph not found")

except GqldbError as e:
    print(f"GQLDB error: {e}")

Complete Example

Python
from gqldb import GqldbClient, GqldbConfig, NodeData, EdgeData, InsertType
from gqldb.types import BulkCreateNodesOptions, BulkCreateEdgesOptions
from gqldb.errors import GqldbError

def main():
    config = GqldbConfig(
        hosts=["localhost:9000"],
        timeout=30
    )

    with GqldbClient(config) as client:
        client.login("admin", "password")
        client.create_graph("dataOpsDemo")
        client.use_graph("dataOpsDemo")

        # Insert nodes
        print("=== Inserting Nodes ===")
        users = [
            NodeData(labels=["User"], properties={"name": "Alice", "age": 30, "active": True}),
            NodeData(labels=["User"], properties={"name": "Bob", "age": 25, "active": True}),
            NodeData(labels=["User"], properties={"name": "Charlie", "age": 35, "active": False}),
            NodeData(labels=["User", "Admin"], properties={"name": "Diana", "age": 28, "active": True}),
        ]

        options = BulkCreateNodesOptions(mode=InsertType.OVERWRITE)
        result = client.insert_nodes("dataOpsDemo", users, options)
        print(f"  Inserted {result.node_count} users")

        # Insert edges
        print("\n=== Inserting Edges ===")
        relationships = [
            EdgeData(label="Follows", from_node_id="u1", to_node_id="u2", properties={"since": "2023-01"}),
            EdgeData(label="Follows", from_node_id="u2", to_node_id="u3", properties={"since": "2023-03"}),
            EdgeData(label="Follows", from_node_id="u1", to_node_id="u4", properties={"since": "2023-06"}),
            EdgeData(label="Knows", from_node_id="u3", to_node_id="u4", properties={"years": 3}),
        ]

        edge_options = BulkCreateEdgesOptions(skip_invalid_nodes=True)
        result = client.insert_edges("dataOpsDemo", relationships, edge_options)
        print(f"  Inserted {result.edge_count} relationships")

        # Verify data
        print("\n=== Current Data ===")
        response = client.gql("MATCH (n:User) RETURN n.name, n.age, n.active ORDER BY n.name")
        for row in response:
            print(f"  {row.get_string(0)}: age={row.get_int(1)}, active={row.get_bool(2)}")

        response = client.gql("MATCH ()-[e]->() RETURN type(e), count(e)")
        for row in response:
            print(f"  {row.get_string(0)}: {row.get_int(1)} edges")

        # Update with overwrite
        print("\n=== Overwrite (Update Existing) ===")
        updated_users = [
            NodeData(labels=["User"], properties={"name": "Alice", "age": 31, "active": True}),  # Update age
            NodeData(labels=["User"], properties={"name": "Eve", "age": 22, "active": True}),    # New user
        ]

        result = client.insert_nodes("dataOpsDemo", updated_users, BulkCreateNodesOptions(mode=InsertType.OVERWRITE))
        print(f"  Upserted {result.node_count} users")

        # Delete inactive users
        print("\n=== Delete Inactive Users ===")
        response = client.delete_nodes_by_condition(
            labels=["User"],
            where="n.active = false"
        )
        print(f"  Deleted {response.rows_affected} inactive users")

        # Delete specific edges
        print("\n=== Delete Old Relationships ===")
        response = client.delete_edges_by_condition(
            label="Follows",
            where="e.since < '2023-04'"
        )
        print(f"  Deleted {response.rows_affected} old relationships")

        # Final state
        print("\n=== Final Data ===")
        response = client.gql("MATCH (n:User) RETURN n.name ORDER BY n.name")
        names = [row.get_string(0) for row in response]
        print(f"  Users: {names}")

        response = client.gql("MATCH ()-[e]->() RETURN count(e)")
        print(f"  Edges: {response.single_int()}")

        # Cleanup
        client.drop_graph("dataOpsDemo")

if __name__ == "__main__":
    main()