Database Operations¶

Learn how to create and configure Grafeo databases in Python.

Creating a Database¶

import grafeo

# In-memory database
db = grafeo.GrafeoDB()

# Persistent database
db = grafeo.GrafeoDB("my_graph.db")

Executing Queries¶

# Execute a query directly on the database
db.execute("INSERT (:Person {name: 'Alix', age: 30})")

# Query and iterate results
result = db.execute("MATCH (p:Person) RETURN p.name, p.age")
for row in result:
    print(row)

Query Timeouts¶

A query may run for 30 seconds by default; past its timeout it fails with GRAFEO-Q003, which can be retried. Set the database's timeout in seconds when you create it, or set timeout for one query:

import math

db = grafeo.GrafeoDB(query_timeout=5)       # every query: at most 5 seconds
db.execute("MATCH (n) RETURN count(n)", timeout=60)        # this query: at most a minute
db.execute("MATCH (n) RETURN count(n)", timeout=math.inf)  # this query: no limit

query_timeout=None or math.inf runs queries without a limit. A timeout of None (the default) keeps the database's, and a timeout that is not a positive number raises ValueError.

Threads¶

Long-running calls release the GIL while Rust works: queries, commits, algorithms, searches, batch writes and upserts, index builds, imports, exports, opening a database and persistence (save(), backups, checkpoints, compact(), close()). Other Python threads keep running meanwhile, and several threads can query one database at once. Cheap calls that read or write one node or edge keep the GIL. A transaction is used from one thread at a time (see Transactions).

Query Languages¶

Grafeo supports multiple query languages:

# GQL (default)
db.execute("MATCH (p:Person) RETURN p")

# Cypher
db.execute_cypher("MATCH (p:Person) RETURN p")

# Gremlin
db.execute_gremlin("g.V().hasLabel('Person')")

# GraphQL
db.execute_graphql("{ Person { name age } }")

# SPARQL (for RDF data)
db.execute_sparql("SELECT ?s ?p ?o WHERE { ?s ?p ?o }")

Transactions¶

Use transactions for atomic operations:

# Begin a transaction
with db.begin_transaction() as tx:
    tx.execute("INSERT (:Person {name: 'Alix'})")
    tx.execute("INSERT (:Person {name: 'Gus'})")
    tx.commit()  # Both inserts committed atomically

# Rollback on error
with db.begin_transaction() as tx:
    tx.execute("INSERT (:Person {name: 'Vincent'})")
    tx.rollback()  # Changes discarded

Direct Node/Edge API¶

Create nodes and edges programmatically:

# Create a node
node = db.create_node(["Person"], {"name": "Alix", "age": 30})
print(f"Created node with ID: {node.id}")

# Create an edge
db.execute("""
    MATCH (a:Person {name: 'Alix'}), (b:Person {name: 'Gus'})
    INSERT (a)-[:KNOWS {since: 2024}]->(b)
""")

Writing many nodes or edges

Each direct call is checked and committed on its own. To write many nodes or edges at once, pass them to one batch call: batch_create_nodes_with_props, batch_create_edges, or upsert_nodes and upsert_edges to insert or update by a key. A batch is checked, written and committed once for all its rows, and creates none of them if one row fails.

ids = db.batch_create_nodes_with_props("Person", [{"name": "Alix"}, {"name": "Gus"}])
db.batch_create_edges([(ids[0], ids[1], "KNOWS", {"since": 2024})])

Label Management¶

Manage node labels directly:

# Add a label to a node
db.add_node_label(node.id, "Employee")

# Remove a label
db.remove_node_label(node.id, "Contractor")

# Get all labels for a node
labels = db.get_node_labels(node.id)
print(labels)  # ['Person', 'Employee']

Admin APIs¶

Inspect and manage the database:

# Database info
info = db.info()
print(f"Nodes: {info['node_count']}, Edges: {info['edge_count']}")

# Detailed statistics
stats = db.detailed_stats()

# Schema information
schema = db.schema()

# Validate database integrity
db.validate()

Named Graphs and Schemas¶

Manage multiple named graphs and schemas within a single database:

# Create and switch between named graphs
db.create_graph("social")
db.create_graph("work")
db.set_graph("social")
print(db.current_graph())  # 'social'
print(db.list_graphs())    # ['social', 'work']
db.reset_graph()           # Back to default graph
social = db.graph("social")  # A handle: works in 'social' without switching
social.execute("MATCH (n) RETURN count(n)")
db.drop_graph("work")

# Schema management
db.set_schema("v1")
print(db.current_schema())  # 'v1'
db.reset_schema()           # Back to default schema

Graph Projections¶

Create filtered virtual views of your graph for algorithms or scoped queries. A projection views the graph set_graph() selects when it is created (the default graph when none is selected) and keeps reading that graph:

# Create a projection with specific labels and edge types
db.create_projection("people",
    node_labels=["Person"],
    edge_types=["KNOWS", "WORKS_WITH"]
)

# Run graph algorithms on the projection only
scores = db.algorithms.pagerank(projection="people", directed=False)
communities = db.algorithms.louvain(projection="people")

print(db.list_projections())  # ['people']
db.drop_projection("people")

Without projection=, db.algorithms reads the graph set_graph() selects, like execute() and CALL grafeo.pagerank(); db.graph("name").algorithms reads one named graph.

key= keys a result by a node property instead of the node id, and makes PageRank, Louvain and label propagation independent of the order the graph was loaded in (see Determinism). Every node in scope (here every node of the projection) needs its own non-null value of the property: a node without one, or two nodes with the same value, raise GrafeoError before the algorithm runs:

scores = db.algorithms.pagerank(projection="people", directed=False, key="id")

Data Import¶

Import CSV and JSON Lines files directly as graph nodes:

# CSV import (each row becomes a node with the given label)
count = db.import_csv("users.csv", "Person", headers=True)
print(f"Imported {count} nodes")

# JSON Lines import (each line is a JSON object)
count = db.import_jsonl("events.jsonl", "Event")

Each column or JSON key becomes a property under the name the file gives it (home city stays home city), and the label is used as given. A CSV without headers (headers=False) names its columns column0, column1, and so on. CSV values are strings; JSON values keep their types. An empty CSV field sets no property.

Loading typed CSV files¶

For large files, load_nodes_csv and load_edges_csv parse the file in Rust, convert each column to its type and write everything in one bulk transaction: if one record cannot be loaded, nothing is.

report = db.load_nodes_csv(
    "people.csv",
    "Person",                      # one label, or a list of labels
    key="id",                      # present in every record, unique in the file
    types={"id": "int64", "name": "string", "born": "date", "notes": "skip"},
)
print(report)  # {'rows': 3, 'created': 3, 'skipped': 0}

db.load_edges_csv(
    "knows.csv",
    "KNOWS",
    source=("Person", "id", "src"),  # the Person whose id equals the record's src
    target=("Person", "id", "dst"),
    types={"src": "int64", "dst": "int64", "since": "int64"},
    missing="skip",                  # or "error" (the default): a key no node has
)

The types are string, int64, float64, bool, date, timestamp and skip (a column that is not stored), in any case; a column types does not name is a string. An empty field, and any field in nulls, sets no property. An edge file's key columns are not stored on the edge, and a key read as text only finds text keys. A file without a header (header=False) takes its column names from the order of types. delimiter sets another separator. Errors name the file's line and column: line 3: 'lots' in column 'score' is no FLOAT64.

The same loads run from any query language, also in grafeo-server and the other bindings:

db.execute(
    "CALL grafeo.load_nodes_csv({path: 'people.csv', labels: ['Person'], key: 'id', "
    "types: {id: 'INT64', born: 'DATE'}}) YIELD rows, created, skipped RETURN *"
)

The loader is new in 0.6.1 and experimental: formats and options may be added.

Change Data Capture¶

Control CDC on a per-transaction basis:

# Enable CDC for a specific transaction
with db.begin_transaction_with_cdc(True) as tx:
    tx.execute("INSERT (:Person {name: 'Alix'})")
    tx.commit()
    # CDC events are captured for this transaction

# Disable CDC for bulk operations
with db.begin_transaction_with_cdc(False) as tx:
    tx.execute("INSERT (:Temp {data: 'scratch'})")
    tx.commit()
    # No CDC events generated

Backup and Restore¶

# Full backup: an image of the database at the current epoch, written while
# commits go on
db.backup_full("/backups")

# Incremental backup (the commits since the last backup), in the same directory
db.backup_incremental("/backups")

# Restore to a specific epoch: the full backup and the incrementals up to it
grafeo.GrafeoDB.restore_to_epoch("/backups", epoch=100, output_path="./restored.grafeo")

An incremental backup needs a full backup in its directory, and continues from the last backup there. The chain survives restarts: once a backup has been taken, close() keeps the write-ahead log written since the last backup (in <file>.wal/), and the next incremental backup after a reopen takes it.

An incremental backup goes to the directory of the last full backup: one to another directory fails with an error naming it, also when a schema statement or graph command came between the two. Backups of one database run one at a time, also to different directories, and a backup directory that another process uses gives a busy error. A backup is on disk before the database lets go of the log it covers. A restore leaves either a complete database or nothing at its output path, so a failed restore can be retried to the same path.

Persistence¶

# Save database to disk
db.save("backup.db")

# Create an in-memory copy (with its indexes and constraints)
memory_db = db.to_memory()

# Load database as in-memory
db = grafeo.GrafeoDB.open_in_memory("my_graph.db")