graphorm 0.3.4

A modern Python ORM for graph databases (RedisGraph/FalkorDB) with type safety, query builder, and automatic property management

GraphORM

GraphORM is a modern Python ORM for graph databases, specifically designed for RedisGraph and FalkorDB. It provides a simple, intuitive API with type safety, automatic property management, and a powerful fluent query builder.

Features

• Type-safe node and edge definitions with Python type hints
• Automatic property management with validation
• Fluent Query Builder API with intuitive syntax
• Transaction support for atomic operations
• Bulk operations for efficient data insertion
• Lazy loading of relationships using Relationship descriptors
• Automatic index creation from class definitions
• Support for explicit labels and relation names
• Batch operations with configurable batch size
• Isolated properties system (separates user properties from internal attributes)

Why GraphORM?

GraphORM makes complex graph queries readable and maintainable. Compare writing raw Cypher with using GraphORM's Query Builder:

Raw Cypher (hard to read and maintain):

query = """
MATCH (a:User {user_id: 1})-[r1:FRIEND]->(b:User)-[r2:FRIEND]->(c:User)
WHERE c.user_id <> 1 AND b.active = true
WITH b, count(r2) as friend_count
WHERE friend_count > 5
MATCH (b)-[r3:FRIEND]->(c:User)
WHERE c.user_id <> 1
RETURN c, friend_count
ORDER BY friend_count DESC
LIMIT 10
"""
result = graph.query(query)

GraphORM Query Builder (readable and type-safe):

from graphorm import select, count

class User(Node):
    __primary_key__ = ["user_id"]
    user_id: int
    active: bool = True

class FRIEND(Edge):
    pass

UserA = User.alias("a")
UserB = User.alias("b")
UserC = User.alias("c")
friend_count_expr = count(FRIEND.alias("r2")).label("friend_count")

stmt = select().match(
    (UserA, FRIEND.alias("r1"), UserB),
    (UserB, FRIEND.alias("r2"), UserC)
).where(
    (UserA.user_id == 1) &
    (UserC.user_id != 1) &
    (UserB.active == True)
).with_(
    UserB,
    friend_count_expr
).where(
    friend_count_expr > 5
).match(
    (UserB, FRIEND.alias("r3"), UserC)
).where(
    UserC.user_id != 1
).returns(
    UserC,
    friend_count_expr
).orderby(
    friend_count_expr.desc()
).limit(10)

result = graph.execute(stmt)

Benefits:

• ✅ Type-safe property access with autocomplete
• ✅ Readable, Pythonic syntax
• ✅ Automatic parameterization (SQL injection safe)
• ✅ Easy to compose and maintain
• ✅ IDE support for refactoring

Requirements

• Python 3.10, 3.11, 3.12, or 3.13
• RedisGraph 2.x or FalkorDB
• Redis server with graph module enabled

Installation

Install from PyPI:

pip install graphorm

Install from source:

git clone https://github.com/hello-tmst/graphorm.git
cd graphorm
pip install -e .

Quick Start

Defining Nodes

from graphorm import Node, Graph

class Page(Node):
    __primary_key__ = ["path"]

    path: str
    parsed: bool = False
    title: str = ""

# Create a graph instance
graph = Graph("my_graph", host="localhost", port=6379)
graph.create()

# Create and add nodes
page = Page(path="/home", parsed=True, title="Home Page")
graph.add_node(page)
graph.flush()

Defining Edges

from graphorm import Edge

class Linked(Edge):
    pass

page1 = Page(path="/page1")
page2 = Page(path="/page2")

graph.add_node(page1)
graph.add_node(page2)

link = Linked(page1, page2)
graph.add_edge(link)
graph.flush()

Inserting Data: From Cypher Pain to Python Joy

Raw Cypher problem	GraphORM solution
String escaping (\", \' in paths, JSON)	Automatic via parameterized queries
Verbose relationship syntax CREATE (a)-[:REL]->(b) per pair	graph.add_edge(Rel(a, b)) with Python objects
Batch insert = huge UNWIND [...] strings	graph.flush(batch_size=1000) — one call, batching under the hood
Type validation only at query execution	Validation at object creation (type hints / property handling)
Manual transaction control	Implicit atomicity: flush() behaves as a single transaction

Before: Raw Cypher

# Insert 3 users + 2 follows relationships
queries = [
    """CREATE (:User {email: "[e-mail]", name: "Alice O\\'Connor", age: 30})""",
    """CREATE (:User {email: "[e-mail]", name: "Bob \\"The Builder\\"", age: 28})""",
    """CREATE (:User {email: "[e-mail]", name: "Carol", age: 35})""",
    """
    MATCH (a:User {email: "[e-mail]"}), (b:User {email: "[e-mail]"})
    CREATE (a)-[:FOLLOWS {since: 1704067200}]->(b)
    """,
    """
    MATCH (b:User {email: "[e-mail]"}), (c:User {email: "[e-mail]"})
    CREATE (b)-[:FOLLOWS {since: 1704067300}]->(c)
    """
]

for q in queries:
    graph.query(q)  # No transaction safety!

Problems:

• Manual escaping for O'Connor, "The Builder"
• 5 separate queries = 5 network round-trips
• No atomicity — if the 4th query fails, the graph is left in a partial state
• No validation — age: "thirty" would only fail at execution time

After: GraphORM

from graphorm import Node, Edge, Graph

class User(Node):
    __primary_key__ = ["email"]
    email: str
    name: str
    age: int

class Follows(Edge):
    since: int = 0

graph = Graph("social", host="localhost", port=6379)
graph.create()

# Create objects — validation happens here
alice = User(email="[e-mail]", name="Alice O'Connor", age=30)
bob = User(email="[e-mail]", name='Bob "The Builder"', age=28)
carol = User(email="[e-mail]", name="Carol", age=35)

# Build relationships — no string interpolation
graph.add_node(alice)
graph.add_node(bob)
graph.add_node(carol)
graph.add_edge(Follows(alice, bob, since=1704067200))
graph.add_edge(Follows(bob, carol, since=1704067300))

# One network call — atomic transaction
graph.flush()

Benefits:

• No escaping — O'Connor and "The Builder" work out of the box
• One network call instead of 5 (batching under the hood)
• Atomic — all or nothing
• Validation before hitting the DB (age="thirty" raises an error at object creation)

Bulk Insert: 10,000 Pages + Links

# Raw Cypher approach (pseudo-code)
cypher = "UNWIND $nodes AS n CREATE (:Page {path: n.path, title: n.title})"
graph.query(cypher, nodes=batch)  # Manual batching, manual error handling

# GraphORM approach (Page and Linked as in Quick Start)
for page_data in pages:
    page = Page(**page_data)
    graph.add_node(page)
    for link_path in page_data.get("links", []):
        graph.add_edge(Linked(page, Page(path=link_path)))

graph.flush(batch_size=1000)  # Automatic batching + rollback on error

Metric	Raw Cypher	GraphORM
Lines of code for 10k nodes	~80	12
Network calls	10 (with manual batching)	1 (automatic batching)
Error handling	Manual per-batch checks	Built-in (rollback on failure)
Development time	2–3 hours	~5 minutes

Query Builder API

GraphORM provides a fluent query builder API:

from graphorm import select, count, indegree, outdegree

# Find unparsed pages
stmt = select().match(Page.alias("p")).where(
    (Page.alias("p").parsed == False) &
    (Page.alias("p").error.is_null())
).limit(10)

result = graph.execute(stmt)

# Find pages with highest degree
total_degree = indegree(Page.alias("p")) + outdegree(Page.alias("p"))
stmt = select().match(Page.alias("p")).where(
    outdegree(Page.alias("p")) > 0
).returns(
    Page.alias("p"),
    total_degree.label("degree")
).orderby(total_degree.desc()).limit(20)

result = graph.execute(stmt)
for row in result.result_set:
    page, degree = row
    print(f"{page.properties['path']}: {degree} connections")

Variable-length paths

You can express variable-length paths (e.g. 1 to 3 hops, unbounded, or exactly N) in ORM style — no raw Cypher — using Edge.variable_length(min_hops, max_hops) in the same match():

from graphorm import Node, Edge, select

class Page(Node):
    __primary_key__ = ["path"]
    path: str

class Linked(Edge):
    pass

# 1 to 3 hops: (start:Page)-[:Linked*1..3]->(end:Page)
stmt = select().match(
    (Page.alias("start"), Linked.variable_length(1, 3), Page.alias("end"))
).returns(Page.alias("start"), Page.alias("end"))

result = graph.execute(stmt)

For unbounded length use Linked.variable_length(); for exactly N hops use Linked.variable_length(2, 2). Raw string patterns in match("(a)-[:Linked*1..3]->(b)") are still supported.

Transactions

Use transactions to group operations atomically:

with graph.transaction() as tx:
    tx.add_node(page1)
    tx.add_node(page2)
    tx.add_edge(Linked(page1, page2))
    # Automatically flushed on exit

Bulk Operations

Efficiently insert large amounts of data using bulk operations:

pages_data = [
    {"path": f"/page{i}", "domain": "example.com", "parsed": False}
    for i in range(10000)
]

result = graph.bulk_upsert(Page, pages_data, batch_size=1000)

Real-World Example: Building a Social Network

Here's a powerful example that demonstrates GraphORM's capabilities for building a social network with thousands of users and connections:

from graphorm import Node, Edge, Graph
import random

# Define the data model
class User(Node):
    __primary_key__ = ["user_id"]
    __indexes__ = ["user_id", "name", "active"]

    user_id: int
    name: str
    email: str
    active: bool = True
    join_date: str = ""

class FRIEND(Edge):
    pass

class FOLLOWS(Edge):
    created_at: str = ""

# Create graph
graph = Graph("social_network", host="localhost", port=6379)
graph.create()

# Generate 10,000 users with bulk insert
users_data = [
    {
        "user_id": i,
        "name": f"User {i}",
        "email": f"user{i}@example.com",
        "active": random.choice([True, False]),
        "join_date": "2024-01-01"
    }
    for i in range(10000)
]

# Bulk insert users (fast and efficient)
print("Creating 10,000 users...")
result = graph.bulk_upsert(User, users_data, batch_size=1000)
print(f"Created {len(users_data)} users in batches")

# Create friendships using transactions
print("Creating 50,000 friendships...")
friendships_created = 0

# Process in batches for better performance
batch_size = 1000
for batch_start in range(0, 50000, batch_size):
    with graph.transaction() as tx:
        for _ in range(batch_size):
            # Randomly connect users
            user1_id = random.randint(0, 9999)
            user2_id = random.randint(0, 9999)

            if user1_id != user2_id:
                user1 = User(user_id=user1_id)
                user2 = User(user_id=user2_id)
                tx.add_edge(FRIEND(user1, user2))
                friendships_created += 1

    if (batch_start // batch_size + 1) % 10 == 0:
        print(f"  Processed {batch_start + batch_size} friendships...")

print(f"Created {friendships_created} friendships")

# Create follow relationships
print("Creating 20,000 follow relationships...")
follows_created = 0

for batch_start in range(0, 20000, batch_size):
    with graph.transaction() as tx:
        for _ in range(batch_size):
            follower_id = random.randint(0, 9999)
            followee_id = random.randint(0, 9999)

            if follower_id != followee_id:
                follower = User(user_id=follower_id)
                followee = User(user_id=followee_id)
                tx.add_edge(FOLLOWS(follower, followee, created_at="2024-01-01"))
                follows_created += 1

print(f"Created {follows_created} follow relationships")

# Query: Find most connected users
from graphorm import select, count, outdegree, indegree

print("\nFinding most connected users...")
UserA = User.alias("u")
stmt = select().match(UserA).where(
    UserA.active == True
).returns(
    UserA,
    (outdegree(UserA) + indegree(UserA)).label("total_connections")
).orderby(
    "total_connections DESC"
).limit(10)

result = graph.execute(stmt)
print("Top 10 most connected users:")
for user, connections in result.result_set:
    print(f"  {user.properties['name']}: {connections} connections")

# Query: Find mutual friends
print("\nFinding mutual friends between two users...")
User1 = User.alias("u1")
User2 = User.alias("u2")
Mutual = User.alias("m")

stmt = select().match(
    (User1, FRIEND.alias("f1"), Mutual),
    (User2, FRIEND.alias("f2"), Mutual)
).where(
    (User1.user_id == 0) & (User2.user_id == 1)
).returns(
    Mutual
).limit(20)

result = graph.execute(stmt)
print(f"Found {len(result.result_set)} mutual friends between User 0 and User 1")

print("\n✅ Social network created successfully!")
print(f"   - {len(users_data)} users")
print(f"   - {friendships_created} friendships")
print(f"   - {follows_created} follow relationships")

What makes this impressive:

• 🚀 10,000 users inserted in seconds using bulk operations
• 🔗 70,000+ relationships created efficiently with transactions
• 📊 Complex queries executed with readable, type-safe syntax
• ⚡ Batch processing for optimal performance
• 🛡️ Atomic operations ensuring data consistency
• 🎯 Type safety throughout - IDE autocomplete works everywhere

This example demonstrates how GraphORM makes it easy to build and query large-scale graph applications with clean, maintainable code.

Relationships

Lazy load related nodes using Relationship descriptors:

from graphorm import Relationship

class Page(Node):
    __primary_key__ = ["path"]
    path: str

    linked_pages = Relationship("Linked", direction="outgoing")
    linked_from = Relationship("Linked", direction="incoming")

page = graph.get_node(Page(path="/home"))
if page:
    for linked_page in page.linked_pages:
        print(linked_page.properties['path'])

Indexes

Automatically create indexes on node properties:

class Page(Node):
    __primary_key__ = ["path"]
    __indexes__ = ["path", "parsed", "domain"]
    path: str
    parsed: bool = False
    domain: str = ""

graph.create()  # Automatically creates indexes from __indexes__

Managing Labels and Relations

Default Behavior

By default, GraphORM uses the class name as-is for labels and relations:

class Page(Node):
    # Label will be "Page" (not "page")
    pass

class MyLink(Edge):
    # Relation will be "MyLink" (not "myLink")
    pass

Explicit Labels and Relations

You can explicitly specify labels and relation names using class attributes:

class CustomPage(Node):
    __label__ = "Page"  # Explicit label
    __primary_key__ = ["path"]
    path: str

class CustomLink(Edge):
    __relation_name__ = "Linked"  # Explicit relation name
    pass

Note: Explicit labels and relations must be non-empty strings. Invalid values will raise a ValueError.

Querying with Labels

GraphORM automatically uses the correct label (class name or explicit label) when building queries. Use the Query Builder API for type-safe, readable queries:

from graphorm import select

# For class Page (label is "Page")
stmt = select().match(Page.alias("p"))
result = graph.execute(stmt)

# For class with explicit label
class MyPage(Node):
    __label__ = "Page"
    __primary_key__ = ["path"]
    path: str

# Query automatically uses "Page" label - GraphORM handles it for you
stmt = select().match(MyPage.alias("p"))
result = graph.execute(stmt)

Note: While you can still use raw Cypher queries with graph.query(), the Query Builder API provides better type safety, readability, and maintainability.

Properties Management

GraphORM includes an isolated properties management system that separates user-defined properties from internal attributes.

Accessing Properties

page = Page(path="/test", parsed=True)

# Get all properties (excluding internal attributes)
props = page.properties
# Returns: {'path': '/test', 'parsed': True}

# Properties are isolated from internal attributes
# Internal attributes like __id__, __alias__, etc. are not included

Updating Properties

# Update properties
page.update({"parsed": False, "title": "New Title"})

# Properties are validated based on type annotations

Examples

Complete Example

from graphorm import Node, Edge, Graph

# Define nodes
class Page(Node):
    __primary_key__ = ["path"]
    path: str
    parsed: bool = False

class Website(Node):
    __primary_key__ = ["domain"]
    domain: str

# Define edges
class Linked(Edge):
    pass

# Create graph
graph = Graph("example", host="localhost", port=6379)
graph.create()

# Create nodes
page1 = Page(path="/page1", parsed=False)
page2 = Page(path="/page2", parsed=True)
website = Website(domain="example.com")

graph.add_node(page1)
graph.add_node(page2)
graph.add_node(website)
graph.flush()

# Create edges
link1 = Linked(page1, page2)
link2 = Linked(page1, website)

graph.add_edge(link1)
graph.add_edge(link2)
graph.flush()

# Query using GraphORM Query Builder
from graphorm import select

# Query for Page targets
PageP = Page.alias("p")
PageTarget = Page.alias("target")
stmt1 = select().match(
    (PageP, Linked.alias("r"), PageTarget)
).where(
    PageP.parsed == False
).returns(
    PageP,
    PageTarget
)

result1 = graph.execute(stmt1)

# Query for Website targets
WebsiteTarget = Website.alias("target")
stmt2 = select().match(
    (PageP, Linked.alias("r"), WebsiteTarget)
).where(
    PageP.parsed == False
).returns(
    PageP,
    WebsiteTarget
)

result2 = graph.execute(stmt2)

# Combine results if needed
all_results = list(result1.result_set) + list(result2.result_set)

# Cleanup
graph.delete()

More Examples

For more detailed examples, see the examples directory:

• Web Crawler - Building a web crawler with GraphORM
• Social Network - Modeling social networks
• Ontology - Working with ontologies

Documentation

• GitHub Repository
• Examples
  • Web Crawler
  • Social Network
  • Ontology

Development

Running Tests

Run tests with coverage:

pytest --cov=graphorm --cov-report=html --cov-report=term-missing

View coverage report:

# HTML report
open htmlcov/index.html

# Terminal report
pytest --cov=graphorm --cov-report=term-missing

Breaking Changes

Version 0.3.0

No breaking changes. This release maintains full backward compatibility with version 0.2.x.

Version 0.2.0+

• Removed dependency on camelcase: Labels and relations now use class names as-is (e.g., Page instead of page)
• All existing code must be updated: Queries using old lowercase labels need to be updated to use class names

Migration Guide

If you're upgrading from an older version:

1. From 0.1.x to 0.2.0+: Update all Cypher queries to use class names instead of camelcase labels:

   # Old (before 0.2.0)
   query = "MATCH (p:page) RETURN p"
   
   # New (0.2.0+)
   query = "MATCH (p:Page) RETURN p"
   

2. If you need to maintain old label names, use explicit labels:

   class Page(Node):
       __label__ = "page"  # Maintain old label
       __primary_key__ = ["path"]
       path: str
   

3. From 0.2.x to 0.3.0: No migration required. All existing code continues to work as before. New features (DELETE, REMOVE, WITH, CASE, etc.) are opt-in.

License

This project is licensed under the MIT License - see the LICENSE file for details.