GraphQLite Python

Python bindings for GraphQLite, a SQLite extension that adds graph database capabilities using Cypher.

Installation

pip install graphqlite

Quick Start

High-Level Graph API (Recommended)

The Graph class provides an ergonomic interface for common graph operations:

from graphqlite import Graph

# Create a graph (in-memory or file-based)
g = Graph(":memory:")

# Add nodes
g.upsert_node("alice", {"name": "Alice", "age": 30}, label="Person")
g.upsert_node("bob", {"name": "Bob", "age": 25}, label="Person")

# Add edges
g.upsert_edge("alice", "bob", {"since": 2020}, rel_type="KNOWS")

# Query
print(g.stats())              # {'nodes': 2, 'edges': 1}
print(g.get_neighbors("alice"))  # [{'id': 'bob', ...}]
print(g.node_degree("alice"))    # 1

# Graph algorithms
ranks = g.pagerank()
communities = g.community_detection()

# Raw Cypher when needed
results = g.query("MATCH (a)-[:KNOWS]->(b) RETURN a.name, b.name")

Low-Level Cypher API

For complex queries or when you need full control:

from graphqlite import connect

db = connect("graph.db")

db.cypher("CREATE (a:Person {name: 'Alice', age: 30})")
db.cypher("CREATE (b:Person {name: 'Bob', age: 25})")
db.cypher("""
    MATCH (a:Person {name: 'Alice'}), (b:Person {name: 'Bob'})
    CREATE (a)-[:KNOWS]->(b)
""")

results = db.cypher("MATCH (a:Person)-[:KNOWS]->(b) RETURN a.name, b.name")
for row in results:
    print(f"{row['a.name']} knows {row['b.name']}")

API Reference

Graph Class

from graphqlite import Graph, graph

# Constructor
g = Graph(db_path=":memory:", namespace="default", extension_path=None)

# Or use the factory function
g = graph(":memory:")

Node Operations

Edge Operations

Graph Queries

Graph Algorithms

Centrality

Community Detection

Connected Components

Path Finding

Traversal

Similarity

Export

Batch Operations

# Batch insert nodes (upsert semantics)
g.upsert_nodes_batch([
    ("n1", {"name": "Alice"}, "Person"),
    ("n2", {"name": "Bob"}, "Person"),
])

# Batch insert edges (upsert semantics)
g.upsert_edges_batch([
    ("n1", "n2", {"weight": 1.0}, "KNOWS"),
])

Bulk Insert (High Performance)

For maximum throughput when building graphs from external data, use the bulk insert methods. These bypass Cypher parsing entirely and use direct SQL, achieving 100-500x faster insert rates.

# Bulk insert nodes - returns dict mapping external_id -> internal_rowid
id_map = g.insert_nodes_bulk([
    ("alice", {"name": "Alice", "age": 30}, "Person"),
    ("bob", {"name": "Bob", "age": 25}, "Person"),
    ("charlie", {"name": "Charlie"}, "Person"),
])

# Bulk insert edges using the ID map - no MATCH queries needed!
edges_inserted = g.insert_edges_bulk([
    ("alice", "bob", {"since": 2020}, "KNOWS"),
    ("bob", "charlie", {"since": 2021}, "KNOWS"),
], id_map)

# Or use the convenience method for both
result = g.insert_graph_bulk(nodes=nodes, edges=edges)
print(f"Inserted {result.nodes_inserted} nodes, {result.edges_inserted} edges")

# Resolve existing node IDs (for edges to pre-existing nodes)
resolved = g.resolve_node_ids(["alice", "bob"])

Connection Class

from graphqlite import connect, wrap

# Open new connection
db = connect("graph.db")
db = connect(":memory:")

# Wrap existing sqlite3 connection
import sqlite3
conn = sqlite3.connect("graph.db")
db = wrap(conn)

Methods

CypherResult

Results from cypher() calls:

results = db.cypher("MATCH (n) RETURN n.name")

len(results)           # Number of rows
results[0]             # First row as dict
results.columns        # Column names
results.to_list()      # All rows as list

for row in results:
    print(row["n.name"])

Utility Functions

from graphqlite import escape_string, sanitize_rel_type, CYPHER_RESERVED

# Escape strings for Cypher queries
safe = escape_string("It's a test")  # "It\\'s a test"

# Sanitize relationship types
rel = sanitize_rel_type("has-items")  # "has_items"
rel = sanitize_rel_type("CREATE")     # "REL_CREATE" (reserved word)

# Set of Cypher reserved keywords
if "MATCH" in CYPHER_RESERVED:
    print("MATCH is reserved")

Extension Path

The extension is located automatically. To specify a custom path:

db = connect("graph.db", extension_path="/path/to/graphqlite.dylib")

Or set the GRAPHQLITE_EXTENSION_PATH environment variable.

Troubleshooting

See FAQ.md for common issues and solutions.

License

MIT