tessl/pypi-neomodel

An object mapper for the neo4j graph database.

—

Pending

Overview

Eval results

Files

Query and Traversal

Name: tessl/pypi-neomodel
Author: tessl

Advanced querying capabilities in neomodel including filtering, ordering, aggregation, and graph traversal operations. Supports complex WHERE clauses, relationship navigation, and both synchronous and asynchronous query patterns.

Capabilities

NodeSet Query Builder

Primary interface for querying and filtering node collections with chainable methods.

class NodeSet:
    """
    Query builder for node collections in synchronous mode.
    
    Provides filtering, ordering, limiting, and aggregation capabilities.
    """
    
    def filter(self, **kwargs):
        """
        Filter nodes by property values and relationships.
        
        Args:
            **kwargs: Filter criteria using Django-style lookups
                     (exact, contains, icontains, startswith, endswith,
                      gt, gte, lt, lte, in, isnull, regex, iregex)
        
        Returns:
            NodeSet: Filtered node collection
        """
    
    def exclude(self, **kwargs):
        """
        Exclude nodes matching criteria.
        
        Args:
            **kwargs: Exclusion criteria using Django-style lookups
        
        Returns:
            NodeSet: Filtered node collection
        """
    
    def get(self, **kwargs):
        """
        Get a single node matching criteria.
        
        Args:
            **kwargs: Filter criteria
        
        Returns:
            Node instance
        
        Raises:
            DoesNotExist: If no node matches
            MultipleNodesReturned: If multiple nodes match
        """
    
    def get_or_none(self, **kwargs):
        """
        Get a single node or None if not found.
        
        Args:
            **kwargs: Filter criteria
        
        Returns:
            Node instance or None
        """
    
    def all(self):
        """
        Get all nodes in the collection.
        
        Returns:
            list: List of node instances
        """
    
    def order_by(self, *properties):
        """
        Order results by properties.
        
        Args:
            *properties: Property names, prefix with '-' for descending
        
        Returns:
            NodeSet: Ordered node collection
        """
    
    def limit(self, count):
        """
        Limit the number of results.
        
        Args:
            count (int): Maximum number of results
        
        Returns:
            NodeSet: Limited node collection
        """
    
    def skip(self, count):
        """
        Skip a number of results.
        
        Args:
            count (int): Number of results to skip
        
        Returns:
            NodeSet: Node collection with offset
        """

class AsyncNodeSet:
    """
    Query builder for node collections in asynchronous mode.
    
    Same interface as NodeSet but all methods are async.
    """
    
    async def filter(self, **kwargs):
        """Asynchronously filter nodes by property values and relationships."""
    
    async def exclude(self, **kwargs):
        """Asynchronously exclude nodes matching criteria."""
    
    async def get(self, **kwargs):
        """Asynchronously get a single node matching criteria."""
    
    async def get_or_none(self, **kwargs):
        """Asynchronously get a single node or None if not found."""
    
    async def all(self):
        """Asynchronously get all nodes in the collection."""
    
    def order_by(self, *properties):
        """Order results by properties (returns modified NodeSet)."""
    
    def limit(self, count):
        """Limit the number of results (returns modified NodeSet)."""
    
    def skip(self, count):
        """Skip a number of results (returns modified NodeSet)."""

Q Objects for Complex Queries

Query objects for building complex logical queries with AND, OR, and NOT operations.

class Q:
    """
    Query object for complex filtering with logical operators.
    
    Supports combining multiple filter conditions with AND, OR, and NOT logic.
    """
    
    def __init__(self, **kwargs):
        """
        Initialize Q object with filter criteria.
        
        Args:
            **kwargs: Filter criteria using Django-style lookups
        """
    
    def __and__(self, other):
        """
        Combine with another Q object using AND logic.
        
        Args:
            other (Q): Another Q object
        
        Returns:
            Q: Combined Q object
        """
    
    def __or__(self, other):
        """
        Combine with another Q object using OR logic.
        
        Args:
            other (Q): Another Q object
        
        Returns:
            Q: Combined Q object
        """
    
    def __invert__(self):
        """
        Negate the Q object (NOT logic).
        
        Returns:
            Q: Negated Q object
        """

Traversal Classes

Classes for advanced graph traversal and path-based operations.

class Traversal:
    """
    Graph traversal operations for synchronous mode.
    
    Enables complex path-based queries and graph navigation.
    """
    
    def traverse(self, *relationships):
        """
        Traverse relationships to connected nodes.
        
        Args:
            *relationships: Relationship definitions to follow
        
        Returns:
            Traversal: Extended traversal path
        """

class AsyncTraversal:
    """
    Graph traversal operations for asynchronous mode.
    
    Same interface as Traversal but for async operations.
    """
    
    def traverse(self, *relationships):
        """
        Traverse relationships to connected nodes asynchronously.
        
        Args:
            *relationships: Relationship definitions to follow
        
        Returns:
            AsyncTraversal: Extended traversal path
        """

Path Classes

Classes representing paths through the graph with node and relationship sequences.

class NeomodelPath:
    """
    Represents Neo4j paths in synchronous mode.
    
    Contains sequences of nodes and relationships forming a path through the graph.
    """
    
    def nodes(self):
        """
        Get nodes in the path.
        
        Returns:
            list: List of node instances in path order
        """
    
    def relationships(self):
        """
        Get relationships in the path.
        
        Returns:
            list: List of relationship instances in path order
        """

class AsyncNeomodelPath:
    """
    Represents Neo4j paths in asynchronous mode.
    
    Same interface as NeomodelPath but for async operations.
    """
    
    def nodes(self):
        """Get nodes in the path."""
    
    def relationships(self):
        """Get relationships in the path."""

Usage Examples

Basic Filtering and Querying

from neomodel import StructuredNode, StringProperty, IntegerProperty

class Person(StructuredNode):
    name = StringProperty(required=True)
    age = IntegerProperty()
    email = StringProperty()

# Basic filtering
adults = Person.nodes.filter(age__gte=18)
johns = Person.nodes.filter(name__icontains='john')
gmail_users = Person.nodes.filter(email__endswith='@gmail.com')

# Exclusion
non_admins = Person.nodes.exclude(email__endswith='@admin.com')

# Ordering and limiting
recent_users = Person.nodes.order_by('-created_at').limit(10)
alphabetical = Person.nodes.order_by('name')

# Get single node
alice = Person.nodes.get(name='Alice')
maybe_bob = Person.nodes.get_or_none(name='Bob')

Complex Queries with Q Objects

from neomodel import Q

# Complex logical conditions
young_or_senior = Person.nodes.filter(
    Q(age__lt=25) | Q(age__gt=65)
)

# Multiple conditions
tech_professionals = Person.nodes.filter(
    Q(email__endswith='@tech.com') & 
    Q(age__gte=25) & 
    Q(age__lte=45)
)

# Negation
non_gmail_adults = Person.nodes.filter(
    ~Q(email__endswith='@gmail.com') & Q(age__gte=18)
)

# Combining with regular filters
filtered_users = Person.nodes.filter(
    Q(name__startswith='A') | Q(name__startswith='B'),
    age__gte=21
)

Advanced Filter Lookups

# String lookups
exact_match = Person.nodes.filter(name='Alice')
contains = Person.nodes.filter(name__contains='ali')
case_insensitive = Person.nodes.filter(name__icontains='ALICE')
starts_with = Person.nodes.filter(name__startswith='Al')
ends_with = Person.nodes.filter(name__endswith='ice')

# Numeric comparisons
young = Person.nodes.filter(age__lt=30)
adults = Person.nodes.filter(age__gte=18)
middle_aged = Person.nodes.filter(age__gt=30, age__lt=50)

# List membership
names = ['Alice', 'Bob', 'Charlie']
selected_people = Person.nodes.filter(name__in=names)

# Null checks
has_email = Person.nodes.filter(email__isnull=False)
no_age = Person.nodes.filter(age__isnull=True)

# Regular expressions
phone_pattern = Person.nodes.filter(phone__regex=r'^\+1\d{10}$')

Async Query Operations

from neomodel import AsyncStructuredNode

class AsyncPerson(AsyncStructuredNode):
    name = StringProperty(required=True)
    age = IntegerProperty()

async def query_people():
    # Async filtering
    adults = await AsyncPerson.nodes.filter(age__gte=18).all()
    alice = await AsyncPerson.nodes.get(name='Alice')
    
    # Async ordering and limiting
    recent = await AsyncPerson.nodes.order_by('-created_at').limit(5).all()
    
    return adults, alice, recent

Relationship-based Queries

class Person(StructuredNode):
    name = StringProperty(required=True)
    friends = RelationshipTo('Person', 'FRIENDS_WITH')
    works_at = RelationshipTo('Company', 'WORKS_AT')

class Company(StructuredNode):
    name = StringProperty(required=True)

# Query through relationships
alice = Person.nodes.get(name='Alice')

# Get all friends
alice_friends = alice.friends.all()

# Filter friends
close_friends = alice.friends.filter(name__contains='Bob')

# Query company employees
company = Company.nodes.get(name='TechCorp')
employees = company.employees.all()

# Complex relationship filtering
tech_workers = Person.nodes.filter(works_at__name='TechCorp')

Graph Traversal Operations

# Traversal through multiple relationship types
alice = Person.nodes.get(name='Alice')

# Traverse friend-of-friend relationships
friends_of_friends = alice.traverse('friends', 'friends').all()

# Complex path traversals
path_results = alice.traverse('works_at', 'employees').filter(age__gt=30)

# Path objects (when supported)
paths = alice.traverse_paths('friends').all()
for path in paths:
    nodes = path.nodes()
    relationships = path.relationships()

Aggregation and Statistics

# Count results
total_users = Person.nodes.count()
adult_count = Person.nodes.filter(age__gte=18).count()

# Check existence
has_admin = Person.nodes.filter(email__endswith='@admin.com').exists()

# First/last
oldest = Person.nodes.order_by('-age').first()
newest = Person.nodes.order_by('-created_at').first()

Types

# Query result types
NodeCollection = List[StructuredNode]
SingleNodeResult = Union[StructuredNode, None]

# Query filter types
FilterLookup = Union[str, int, float, bool, list]
QueryDict = Dict[str, FilterLookup]

# Traversal types
PathResult = Union[NeomodelPath, AsyncNeomodelPath]
TraversalResult = Union[NodeCollection, PathResult]

Install with Tessl CLI