tessl/pypi-neomodel
An object mapper for the neo4j graph database.
—
Pending
nodes.mddocs/
Node Management
Core functionality for defining and working with graph nodes in neomodel. This includes model definitions, CRUD operations, property management, and both synchronous and asynchronous implementations.
Capabilities
Structured Node Classes
Base classes for defining graph node models with property definitions, relationship management, and database operations.
class StructuredNode:
"""
Base class for all node models in synchronous mode.
Provides CRUD operations, querying, and relationship management.
"""
def save(self):
"""
Save the node to Neo4j database.
Returns:
self: The saved node instance
"""
def delete(self):
"""
Delete the node from Neo4j database.
Returns:
bool: True if deletion was successful
"""
def refresh(self):
"""
Reload the node's properties from the database.
Returns:
self: The refreshed node instance
"""
@classmethod
def create(cls, **kwargs):
"""
Create a new node instance and save it to the database.
Args:
**kwargs: Property values for the new node
Returns:
Instance of the node class
"""
@classmethod
def get_or_create(cls, defaults=None, **kwargs):
"""
Get an existing node or create a new one if it doesn't exist.
Args:
defaults (dict, optional): Default values for creation
**kwargs: Lookup parameters
Returns:
tuple: (node_instance, created_boolean)
"""
@classmethod
def create_or_update(cls, **kwargs):
"""
Create a new node or update existing one based on unique properties.
Args:
**kwargs: Property values
Returns:
Instance of the node class
"""
class AsyncStructuredNode:
"""
Base class for all node models in asynchronous mode.
Provides async CRUD operations, querying, and relationship management.
"""
async def save(self):
"""
Asynchronously save the node to Neo4j database.
Returns:
self: The saved node instance
"""
async def delete(self):
"""
Asynchronously delete the node from Neo4j database.
Returns:
bool: True if deletion was successful
"""
async def refresh(self):
"""
Asynchronously reload the node's properties from the database.
Returns:
self: The refreshed node instance
"""
@classmethod
async def create(cls, **kwargs):
"""
Asynchronously create a new node instance and save it to the database.
Args:
**kwargs: Property values for the new node
Returns:
Instance of the node class
"""
@classmethod
async def get_or_create(cls, defaults=None, **kwargs):
"""
Asynchronously get an existing node or create a new one.
Args:
defaults (dict, optional): Default values for creation
**kwargs: Lookup parameters
Returns:
tuple: (node_instance, created_boolean)
"""
@classmethod
async def create_or_update(cls, **kwargs):
"""
Asynchronously create a new node or update existing one.
Args:
**kwargs: Property values
Returns:
Instance of the node class
"""Property Management
Classes responsible for property handling, validation, and data transformation between Python and Neo4j formats.
class PropertyManager:
"""
Base class for property management in synchronous mode.
Handles property inflation/deflation, validation, and database interaction.
"""
class AsyncPropertyManager:
"""
Base class for property management in asynchronous mode.
Handles async property inflation/deflation, validation, and database interaction.
"""Usage Examples
Basic Node Definition
from neomodel import StructuredNode, StringProperty, IntegerProperty
class Person(StructuredNode):
name = StringProperty(unique_index=True, required=True)
age = IntegerProperty()
email = StringProperty()
# Create and save a node
person = Person(name="Alice", age=30, email="alice@example.com").save()
# Alternative creation method
person = Person.create(name="Bob", age=25)
# Get or create pattern
person, created = Person.get_or_create(
name="Charlie",
defaults={'age': 35, 'email': 'charlie@example.com'}
)Async Node Operations
from neomodel import AsyncStructuredNode, StringProperty, IntegerProperty
class AsyncPerson(AsyncStructuredNode):
name = StringProperty(unique_index=True, required=True)
age = IntegerProperty()
# Async operations
async def create_person():
person = await AsyncPerson.create(name="David", age=40)
await person.save()
# Get or create async
person, created = await AsyncPerson.get_or_create(
name="Eve",
defaults={'age': 28}
)
return personNode Querying
# Query through the nodes manager
all_people = Person.nodes.all()
alice = Person.nodes.get(name="Alice")
adults = Person.nodes.filter(age__gte=18)
# Async querying
async def query_people():
all_people = await AsyncPerson.nodes.all()
alice = await AsyncPerson.nodes.get(name="Alice")
return all_people, aliceSemi-structured Nodes
Special node classes that allow storing properties not defined in the class schema alongside the defined properties.
class SemiStructuredNode(StructuredNode):
"""
Base class for synchronous semi-structured nodes.
Allows properties to be stored on a node that aren't specified in its
definition. Conflicting properties with class attributes raise DeflateConflict.
"""
__abstract_node__ = True
@classmethod
def inflate(cls, node): ...
@classmethod
def deflate(cls, node_props, obj=None, skip_empty=False): ...
class AsyncSemiStructuredNode(AsyncStructuredNode):
"""
Base class for asynchronous semi-structured nodes.
Allows properties to be stored on a node that aren't specified in its
definition. Conflicting properties with class attributes raise DeflateConflict.
"""
__abstract_node__ = True
@classmethod
def inflate(cls, node): ...
@classmethod
def deflate(cls, node_props, obj=None, skip_empty=False): ...Semi-structured Usage Example
from neomodel import StringProperty, IntegerProperty
from neomodel.contrib.sync_.semi_structured import SemiStructuredNode
from neomodel.contrib.async_.semi_structured import AsyncSemiStructuredNode
class Person(SemiStructuredNode):
name = StringProperty(required=True)
age = IntegerProperty()
# Create node with extra properties not in schema
person = Person(
name='Tim',
age=25,
weight=70, # Extra property
occupation='Engineer' # Extra property
).save()
# Access both defined and extra properties
print(person.name) # Defined property
print(person.weight) # Extra property
print(person.occupation) # Extra property
# Adding conflicting properties will raise DeflateConflict
person.name = "Updated Name" # OK - defined property
# person.save = "invalid" # Would raise DeflateConflict (conflicts with method)Types
# Node operation return types
NodeInstance = Union[StructuredNode, AsyncStructuredNode]
CreateResult = Tuple[NodeInstance, bool] # (instance, was_created)Install with Tessl CLI
npx tessl i tessl/pypi-neomodel