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 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)
""")
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
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:
# Create a projection with specific labels and edge types
db.create_projection("people",
node_labels=["Person"],
edge_types=["KNOWS", "WORKS_WITH"]
)
print(db.list_projections()) # ['people']
db.drop_projection("people")
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")
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
db.backup_full("/backups/full")
# Incremental backup (WAL records since last backup)
db.backup_incremental("/backups/incr")
# Restore to a specific epoch
grafeo.GrafeoDB.restore_to_epoch("/backups/full", epoch=100, output_path="./restored")