tessl/pypi-neomodel

An object mapper for the neo4j graph database.

—

Pending

Overview

Eval results

Files

Relationship Management

Name: tessl/pypi-neomodel
Author: tessl

Powerful relationship definition and traversal system in neomodel with cardinality constraints, supporting both simple connections and complex relationship models with properties. Includes both synchronous and asynchronous implementations.

Capabilities

Relationship Definition Classes

Classes for defining different types of relationships between nodes with cardinality and direction constraints.

class Relationship:
    """
    Basic relationship definition for synchronous operations.
    
    Args:
        cls_name (str): Target node class name
        relation_type (str): Neo4j relationship type
        **kwargs: Additional relationship configuration
    """
    def __init__(self, cls_name, relation_type, **kwargs): ...

class RelationshipTo(Relationship):
    """
    Outgoing relationship definition for synchronous operations.
    
    Args:
        cls_name (str): Target node class name
        relation_type (str): Neo4j relationship type
        cardinality (class, optional): Cardinality constraint class
        model (class, optional): StructuredRel class for relationship properties
        **kwargs: Additional relationship configuration
    """
    def __init__(self, cls_name, relation_type, cardinality=None, model=None, **kwargs): ...

class RelationshipFrom(Relationship):
    """
    Incoming relationship definition for synchronous operations.
    
    Args:
        cls_name (str): Source node class name
        relation_type (str): Neo4j relationship type
        cardinality (class, optional): Cardinality constraint class
        model (class, optional): StructuredRel class for relationship properties
        **kwargs: Additional relationship configuration
    """
    def __init__(self, cls_name, relation_type, cardinality=None, model=None, **kwargs): ...

class RelationshipDefinition:
    """
    Generic relationship configuration class for synchronous operations.
    
    Args:
        relation_type (str): Neo4j relationship type
        cls_name (str): Target/source node class name
        direction (int): Relationship direction (OUTGOING, INCOMING, EITHER)
        **kwargs: Additional configuration
    """
    def __init__(self, relation_type, cls_name, direction, **kwargs): ...

# Async relationship classes
class AsyncRelationship:
    """Basic relationship definition for asynchronous operations."""
    def __init__(self, cls_name, relation_type, **kwargs): ...

class AsyncRelationshipTo(AsyncRelationship):
    """Outgoing relationship definition for asynchronous operations."""
    def __init__(self, cls_name, relation_type, cardinality=None, model=None, **kwargs): ...

class AsyncRelationshipFrom(AsyncRelationship):
    """Incoming relationship definition for asynchronous operations."""
    def __init__(self, cls_name, relation_type, cardinality=None, model=None, **kwargs): ...

class AsyncRelationshipDefinition:
    """Generic relationship configuration class for asynchronous operations."""
    def __init__(self, relation_type, cls_name, direction, **kwargs): ...

Structured Relationship Classes

Base classes for creating relationship models with properties.

class StructuredRel:
    """
    Base class for relationship models with properties in synchronous mode.
    
    Provides property management and CRUD operations for relationships.
    """
    
    def save(self):
        """
        Save relationship properties to Neo4j database.
        
        Returns:
            self: The saved relationship instance
        """
    
    def delete(self):
        """
        Delete the relationship from Neo4j database.
        
        Returns:
            bool: True if deletion was successful
        """

class AsyncStructuredRel:
    """
    Base class for relationship models with properties in asynchronous mode.
    
    Provides async property management and CRUD operations for relationships.
    """
    
    async def save(self):
        """
        Asynchronously save relationship properties to Neo4j database.
        
        Returns:
            self: The saved relationship instance
        """
    
    async def delete(self):
        """
        Asynchronously delete the relationship from Neo4j database.
        
        Returns:
            bool: True if deletion was successful
        """

Relationship Management Classes

Classes that handle relationship collections and operations on node instances.

class RelationshipManager:
    """
    Manages relationship collections for synchronous operations.
    
    Provides methods for connecting, disconnecting, and querying relationships.
    """
    
    def connect(self, node, properties=None):
        """
        Create a relationship to another node.
        
        Args:
            node: Target node instance
            properties (dict, optional): Relationship properties
        
        Returns:
            Relationship instance or bool
        """
    
    def disconnect(self, node):
        """
        Remove relationship to another node.
        
        Args:
            node: Target node instance
        
        Returns:
            bool: True if disconnection was successful
        """
    
    def all(self):
        """
        Get all related nodes.
        
        Returns:
            list: List of related node instances
        """
    
    def get(self, **kwargs):
        """
        Get a single related node matching criteria.
        
        Args:
            **kwargs: Filter criteria
        
        Returns:
            Node instance
        """
    
    def filter(self, **kwargs):
        """
        Filter related nodes by criteria.
        
        Args:
            **kwargs: Filter criteria
        
        Returns:
            NodeSet: Filtered node collection
        """

class AsyncRelationshipManager:
    """
    Manages relationship collections for asynchronous operations.
    """
    
    async def connect(self, node, properties=None):
        """Asynchronously create a relationship to another node."""
    
    async def disconnect(self, node):
        """Asynchronously remove relationship to another node."""
    
    async def all(self):
        """Asynchronously get all related nodes."""
    
    async def get(self, **kwargs):
        """Asynchronously get a single related node matching criteria."""
    
    async def filter(self, **kwargs):
        """Asynchronously filter related nodes by criteria."""

Cardinality Classes

Classes that enforce relationship cardinality constraints.

# Synchronous cardinality classes
class One:
    """
    Exactly one relationship constraint.
    
    Ensures that exactly one relationship of this type exists.
    """

class ZeroOrOne:
    """
    Zero or one relationship constraint.
    
    Allows at most one relationship of this type.
    """

class OneOrMore:
    """
    One or more relationships constraint.
    
    Requires at least one relationship of this type.
    """

class ZeroOrMore:
    """
    Zero or more relationships constraint.
    
    No limit on the number of relationships (default behavior).
    """

# Asynchronous cardinality classes
class AsyncOne:
    """Exactly one relationship constraint for async operations."""

class AsyncZeroOrOne:
    """Zero or one relationship constraint for async operations."""

class AsyncOneOrMore:
    """One or more relationships constraint for async operations."""

class AsyncZeroOrMore:
    """Zero or more relationships constraint for async operations."""

Usage Examples

Basic Relationship Definition

from neomodel import (
    StructuredNode, StringProperty, RelationshipTo, 
    RelationshipFrom, ZeroOrMore, One
)

class Person(StructuredNode):
    name = StringProperty(required=True)
    
    # Outgoing relationships
    friends = RelationshipTo('Person', 'FRIENDS_WITH', cardinality=ZeroOrMore)
    works_at = RelationshipTo('Company', 'WORKS_AT', cardinality=One)
    
    # Incoming relationships
    managed_by = RelationshipFrom('Person', 'MANAGES', cardinality=ZeroOrMore)

class Company(StructuredNode):
    name = StringProperty(required=True)
    
    # Employees working at this company
    employees = RelationshipFrom('Person', 'WORKS_AT', cardinality=ZeroOrMore)

# Create nodes and relationships
alice = Person(name="Alice").save()
bob = Person(name="Bob").save()
company = Company(name="TechCorp").save()

# Connect nodes
alice.friends.connect(bob)
alice.works_at.connect(company)

Relationship Properties

from neomodel import (
    StructuredNode, StructuredRel, StringProperty, 
    DateTimeProperty, RelationshipTo
)

class FriendshipRel(StructuredRel):
    """Relationship model with properties."""
    since = DateTimeProperty(default_now=True)
    strength = StringProperty(choices=['weak', 'strong'], default='weak')

class Person(StructuredNode):
    name = StringProperty(required=True)
    
    # Relationship with properties
    friends = RelationshipTo(
        'Person', 
        'FRIENDS_WITH', 
        model=FriendshipRel,
        cardinality=ZeroOrMore
    )

# Create friendship with properties
alice = Person(name="Alice").save()
bob = Person(name="Bob").save()

# Connect with relationship properties
friendship = alice.friends.connect(bob, {'strength': 'strong'})
friendship.since = datetime.now()
friendship.save()

Async Relationship Operations

from neomodel import (
    AsyncStructuredNode, AsyncStructuredRel, StringProperty,
    AsyncRelationshipTo, AsyncZeroOrMore
)

class AsyncFriendshipRel(AsyncStructuredRel):
    strength = StringProperty(default='weak')

class AsyncPerson(AsyncStructuredNode):
    name = StringProperty(required=True)
    friends = AsyncRelationshipTo(
        'AsyncPerson', 
        'FRIENDS_WITH',
        model=AsyncFriendshipRel,
        cardinality=AsyncZeroOrMore
    )

async def create_friendship():
    alice = await AsyncPerson(name="Alice").save()
    bob = await AsyncPerson(name="Bob").save()
    
    # Async relationship operations
    await alice.friends.connect(bob, {'strength': 'strong'})
    friends = await alice.friends.all()
    
    return friends

Relationship Traversal and Filtering

# Query through relationships
alice_friends = alice.friends.all()
close_friends = alice.friends.filter(strength='strong')
work_colleagues = alice.works_at.get().employees.all()

# Disconnect relationships
alice.friends.disconnect(bob)

# Check if relationship exists
is_friend = bob in alice.friends.all()

# Get relationship properties
for friend in alice.friends.all():
    rel = alice.friends.relationship(friend)
    print(f"Friends since: {rel.since}")

Complex Relationship Patterns

from neomodel import OUTGOING, INCOMING, EITHER

class Person(StructuredNode):
    name = StringProperty(required=True)
    
    # Bidirectional relationship
    knows = Relationship('Person', 'KNOWS', direction=EITHER)
    
    # Self-referencing with different roles
    manages = RelationshipTo('Person', 'MANAGES')
    managed_by = RelationshipFrom('Person', 'MANAGES')

# Traverse in different directions
manager = person.managed_by.get()
subordinates = person.manages.all()
mutual_connections = person.knows.all()

Types

# Relationship direction constants
OUTGOING: int  # Value: 1
INCOMING: int  # Value: -1  
EITHER: int    # Value: 0

# Cardinality constraint types
CardinalityClass = Union[One, ZeroOrOne, OneOrMore, ZeroOrMore]
AsyncCardinalityClass = Union[AsyncOne, AsyncZeroOrOne, AsyncOneOrMore, AsyncZeroOrMore]

# Relationship operation return types
RelationshipInstance = Union[StructuredRel, AsyncStructuredRel]
ConnectionResult = Union[RelationshipInstance, bool]

Install with Tessl CLI