falkordb-orm 1.2.0

Object-Graph Mapping for FalkorDB with Spring Data-inspired patterns

FalkorDB Python ORM

Object-Graph Mapping for FalkorDB with Spring Data-inspired patterns

FalkorDB Python ORM provides intuitive, annotation-based object-graph mapping for FalkorDB, enabling developers to work with graph databases using familiar ORM patterns inspired by Spring Data.

🎯 Project Status

Production Ready! ✅ The FalkorDB Python ORM is fully implemented, tested, and documented.

🚀 Features

Core Capabilities

• 🏷️ Decorator-based Entity Mapping: Use @node and property decorators for intuitive object-graph mapping
• 📦 Repository Pattern: Complete CRUD operations with type-safe Repository[T]
• 🔑 ID Management: Auto-generated or manual IDs with generated_id()
• 🔄 Type Conversion: Built-in converters for common Python types
• 🎯 Multiple Labels: Support for multiple node labels per entity
• 🎨 Type Safety: Full type hints and generic repositories for IDE support

Query Features

• 🔍 Derived Query Methods: Auto-generate queries from method names (e.g., find_by_name(), count_by_age_greater_than())
• 📊 Comparison Operators: 14+ operators (equals, greater_than, less_than, between, in, etc.)
• 🔗 Logical Operators: AND/OR combinations in queries
• 📝 String Operations: CONTAINS, STARTS WITH, ENDS WITH, regex patterns
• 📊 Sorting & Limiting: ORDER BY multiple fields, first/top_N result limiting
• ⚡ Query Caching: Automatic QuerySpec caching for performance
• 📝 Custom Cypher Queries: @query decorator with parameter binding
• 📊 Aggregation Methods: Built-in sum(), avg(), min(), max() functions

Relationships

• 🔗 Relationship Declaration: relationship() decorator with full type support
• 💤 Lazy Loading: Relationships loaded on-demand with automatic caching
• ⚡ Eager Loading: Solve N+1 queries with fetch=['rel1', 'rel2'] parameter
• 🔄 Cascade Operations: Auto-save related entities with cascade=True
• ↔️ Bidirectional Relationships: Full support for complex relationship graphs
• 🔙 Reverse Relationships: direction='INCOMING' for inverse traversal
• 🔁 Circular Reference Handling: Safe handling of circular relationships

Async Support

• ⚡ AsyncRepository: Full async/await support for all CRUD operations
• 🔄 Async Relationships: Async lazy loading with AsyncLazyList and AsyncLazySingle
• 📊 Async Derived Queries: Auto-generated async query methods
• 🌐 Framework Ready: Perfect for FastAPI, aiohttp, and async Python applications

Advanced Features (v1.1.0 NEW! 🎉)

• ⚡ Transaction Support: Context managers with identity map and change tracking
• 🗂️ Index Management: @indexed and @unique decorators with schema validation
• 📄 Pagination: Full pagination with sorting and navigation (Pageable, Page[T])
• 🔄 Relationship Updates: Automatic deletion of old edges when relationships change

Production Features

• 📚 Comprehensive Documentation: Complete API reference and migration guides
• 🧠 Enhanced Exceptions: Contextual error messages with structured error information
• 🚀 CI/CD Workflows: Automated testing, linting, and publishing
• 💾 Memory Optimization: Interned strings for repeated values with @interned decorator
• 🧪 Integration Tests: Full end-to-end tests with real FalkorDB

📋 Future Enhancements (Optional)

• 📦 Migration System: Schema version management and migrations
• 🔍 Query Result Caching: Result caching for performance
• ⚙️ Batch Optimization: UNWIND-based bulk operations

📜 Usage

Entity Definition

from falkordb_orm import node, property, relationship, Repository
from typing import Optional, List

@node(labels=["Person", "Individual"])
class Person:
    id: Optional[int] = None
    name: str = property("full_name")  # Maps to 'full_name' in graph
    email: str
    age: int
    
    friends: List["Person"] = relationship(type="KNOWS", direction="OUTGOING")
    company: Optional["Company"] = relationship(type="WORKS_FOR", direction="OUTGOING")

@node("Company")
class Company:
    id: Optional[int] = None
    name: str
    employees: List[Person] = relationship(type="WORKS_FOR", direction="INCOMING")

Repository Usage

from falkordb import FalkorDB

# Connect to FalkorDB
db = FalkorDB(host='localhost', port=6379)
graph = db.select_graph('social')

# Create repository
repo = Repository(graph, Person)

# Create and save
person = Person(name="Alice", email="[e-mail]", age=25)
saved = repo.save(person)

# Derived queries (auto-implemented)
adults = repo.find_by_age_greater_than(18)
alice = repo.find_by_name("Alice")
count = repo.count_by_age(25)
exists = repo.exists_by_email("[e-mail]")

# Eager loading relationships (prevents N+1 queries)
person_with_friends = repo.find_by_id(1, fetch=["friends", "company"])
all_with_friends = repo.find_all(fetch=["friends"])  # Single query!

# Cascade save (auto-saves related entities)
company = Company(name="Acme Corp")
employee = Employee(name="Bob", company=company)
repo.save(employee)  # Company automatically saved!

Async Usage (Phase 5)

import asyncio
from falkordb.asyncio import FalkorDB
from falkordb_orm import node, AsyncRepository
from typing import Optional

@node("Person")
class Person:
    id: Optional[int] = None
    name: str
    age: int

async def main():
    # Connect to FalkorDB with async client
    from redis.asyncio import BlockingConnectionPool
    pool = BlockingConnectionPool(max_connections=16, timeout=None, decode_responses=True)
    db = FalkorDB(connection_pool=pool)
    graph = db.select_graph('social')
    
    # Create async repository
    repo = AsyncRepository(graph, Person)
    
    # All operations are async
    person = Person(name="Alice", age=25)
    saved = await repo.save(person)
    
    # Async derived queries
    adults = await repo.find_by_age_greater_than(18)
    count = await repo.count()
    
    # Async eager loading
    person_with_friends = await repo.find_by_id(1, fetch=["friends"])
    
    print(f"Found {count} people")

asyncio.run(main())

Transaction Support (v1.1.0 NEW!)

from falkordb_orm import Session

# Use session for transactions with identity map
with Session(graph) as session:
    # Get entity (cached in identity map)
    person = session.get(Person, 1)
    person.age = 31
    session._dirty.add(person)  # Mark as modified
    
    # Add new entities
    new_person = Person(name="Bob", age=25)
    session.add(new_person)
    
    # Auto-commit on success, auto-rollback on error
    session.commit()

Index Management (v1.1.0 NEW!)

from falkordb_orm import node, indexed, unique, IndexManager

@node("User")
class User:
    email: str = unique(required=True)       # Unique constraint
    age: int = indexed()                      # Regular index
    bio: str = indexed(index_type="FULLTEXT") # Full-text search

# Create indexes
manager = IndexManager(graph)
manager.create_indexes(User, if_not_exists=True)

# Schema validation
from falkordb_orm import SchemaManager
schema_manager = SchemaManager(graph)
result = schema_manager.validate_schema([User, Product])
if not result.is_valid:
    schema_manager.sync_schema([User, Product])

Pagination (v1.1.0 NEW!)

from falkordb_orm import Pageable

# Create pageable (page 0, 10 items, sorted by age)
pageable = Pageable(page=0, size=10, sort_by="age", direction="ASC")

# Get paginated results
page = repo.find_all_paginated(pageable)

print(f"Page {page.page_number + 1} of {page.total_pages}")
print(f"Total: {page.total_elements} items")

for person in page.content:
    print(person.name)

# Navigate pages
if page.has_next():
    next_page = repo.find_all_paginated(pageable.next())

Getting Started

For a complete walkthrough, see QUICKSTART.md.

Custom Queries (Phase 4)

from falkordb_orm import query

class PersonRepository(Repository[Person]):
    @query(
        "MATCH (p:Person)-[:KNOWS]->(f:Person) WHERE p.name = $name RETURN f",
        returns=Person
    )
    def find_friends(self, name: str) -> List[Person]:
        pass
    
    @query(
        "MATCH (p:Person) WHERE p.age BETWEEN $min AND $max RETURN p",
        returns=Person
    )
    def find_by_age_range(self, min: int, max: int) -> List[Person]:
        pass

📋 Comparison with Current Approach

Before (Raw Cypher)

from falkordb import FalkorDB

db = FalkorDB(host='localhost', port=6379)
g = db.select_graph('social')

# Create manually
g.query('''
    CREATE (p:Person {name: $name, email: $email, age: $age})
    RETURN p, id(p)
''', {'name': 'Alice', 'email': '[e-mail]', 'age': 25})

# Query manually
result = g.query('''
    MATCH (p:Person) WHERE p.age > $min_age RETURN p
''', {'min_age': 18})

# Manual result processing
for record in result.result_set:
    person = record[0]
    print(person.properties['name'])

After (ORM - Planned)

from falkordb import FalkorDB
from falkordb_orm import node, Repository

@node("Person")
class Person:
    id: Optional[int] = None
    name: str
    email: str
    age: int

db = FalkorDB(host='localhost', port=6379)
graph = db.select_graph('social')
repo = Repository(graph, Person)

# Create with ORM
person = Person(name="Alice", email="[e-mail]", age=25)
saved = repo.save(person)

# Query with derived method
adults = repo.find_by_age_greater_than(18)

# Automatic object mapping
for person in adults:
    print(person.name)

📚 Documentation

• Quick Start Guide - Get started in minutes
• Design Document - Comprehensive design and architecture
• API Reference - Complete API documentation
  • Decorators - @node, property(), relationship()
  • Repository - Repository and AsyncRepository
• Migration Guide - Migrating from raw FalkorDB client
• Examples - Complete working examples
• Contributing - Contribution guidelines

🤝 Comparison with Spring Data FalkorDB

This project is inspired by Spring Data FalkorDB, bringing similar patterns to Python:

Feature	Spring Data FalkorDB	falkordb-orm
Entity Mapping	@Node annotation	@node decorator
Property Mapping	@Property	property() function
Relationships	@Relationship	relationship() function
Repository	FalkorDBRepository<T, ID>	Repository[T]
Query Methods	Method name derivation	Method name derivation
Custom Queries	@Query annotation	@query decorator
Transactions	@Transactional	Session (unit of work)
Async Support	❌	✅ Planned

🎯 Goals

1. Developer Productivity: Reduce boilerplate code and manual Cypher writing
2. Type Safety: Leverage Python type hints for better IDE support
3. Familiarity: Use patterns developers know from SQLAlchemy, Django ORM, Spring Data
4. Performance: Optimize queries, support batching, implement lazy loading
5. Flexibility: Support both simple and complex use cases

🔧 Technology Stack

• Python: 3.8+
• FalkorDB Client: falkordb-py
• Type Hints: For IDE support and validation
• Decorators: For entity and query definition
• Generics: For type-safe repositories

📦 Installation

pip install falkordb-orm

Or install from source:

git clone https://github.com/FalkorDB/falkordb-py-orm.git
cd falkordb-py-orm
pip install -e .

💡 Design Principles

1. Convention over Configuration: Sensible defaults with customization options
2. Explicit is Better: Clear, readable API over implicit magic
3. Performance Matters: Optimize for common use cases
4. Pythonic: Follow Python idioms and best practices
5. Compatibility: Work seamlessly with existing falkordb-py code

📄 License

MIT License - See LICENSE file for details

🙏 Acknowledgments

• Inspired by Spring Data FalkorDB
• Built on top of falkordb-py
• Follows patterns from SQLAlchemy and Django ORM

🤝 Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

📦 Resources

• FalkorDB: falkordb.com
• Discord: FalkorDB Community
• GitHub: FalkorDB Organization