How to Choose Between Async and Sync Neomodel APIs: A Complete Guide

Choose the synchronous neomodel API for traditional Flask or Django applications and simple scripts, and select the asynchronous API when building high-concurrency services with FastAPI or handling hundreds of simultaneous database connections.

The neo4j-contrib/neomodel library provides dual Python APIs that share identical model definitions but differ fundamentally in execution models. When you choose between async and sync neomodel APIs, you are selecting between blocking I/O operations that execute sequentially and non-blocking coroutines that enable high-concurrency database access. Both implementations maintain complete feature parity, allowing you to switch between them by updating imports and adding await keywords.

Understanding the Dual API Architecture

Neomodel maintains two parallel codebases under separate package paths to support both execution models without mixing blocking and non-blocking code.

Core Class Differences

The synchronous API centers on StructuredNode imported from neomodel, while the asynchronous API uses AsyncStructuredNode from neomodel.async_. In neomodel/sync_/node.py, the StructuredNode.save() method operates as a regular function that blocks until the database responds, whereas in neomodel/async_/node.py, AsyncStructuredNode.save() is defined as async def and must be awaited to execute.

Source Code Organization

The library organizes implementations into mirrored directories with identical method signatures:

• neomodel/sync_/node.py defines StructuredNode with blocking CRUD operations
• neomodel/async_/node.py defines AsyncStructuredNode with coroutine-based methods
• neomodel/sync_/database.py and neomodel/async_/database.py provide blocking and async database wrappers respectively
• neomodel/sync_/match.py offers NodeSet for synchronous queries, while neomodel/async_/match.py provides AsyncNodeSet for async iteration
• neomodel/sync_/transaction.py and neomodel/async_/transaction.py handle transaction context managers

When to Use the Synchronous Neomodel API

Select the synchronous API when your application architecture favors blocking execution or integrates with legacy codebases that do not support async/await patterns.

• Traditional web frameworks: Flask, Django, and Pyramid applications without async support should use StructuredNode to avoid complexity and thread-safety issues.
• Simple scripts and CLI tools: When writing data migration scripts, administrative commands, or ETL jobs, the synchronous API eliminates asyncio boilerplate and event loop management.
• Blocking third-party libraries: If your workflow depends on pandas, standard CSV modules, or other blocking libraries, mixing async and sync code can cause deadlocks unless carefully managed with thread pools.
• Low concurrency requirements: For applications handling fewer than approximately 100 simultaneous database interactions, the overhead of an event loop provides no meaningful benefit while adding cognitive complexity.

Example: Synchronous model definition and query

from neomodel import StructuredNode, StringProperty, IntegerProperty

class Person(StructuredNode):
    name = StringProperty()
    age = IntegerProperty()

# Blocking save operation

john = Person(name="John", age=30).save()

# Synchronous iteration

people = Person.nodes.filter(age__gt=20).all()
for person in people:
    print(person.name, person.age)

When to Use the Asynchronous Neomodel API

Choose the asynchronous API for high-concurrency services and modern async Python frameworks where non-blocking I/O is essential for performance.

• Async web frameworks: FastAPI, Sanic, aiohttp, and Quart applications should use AsyncStructuredNode to maintain non-blocking request handling throughout the entire stack.
• High concurrency workloads: When handling hundreds of simultaneous database connections, the async API multiplexes requests over a single connection pool without blocking threads, enabling significantly higher throughput.
• Streaming large result sets: The AsyncNodeSet in neomodel/async_/match.py provides async for iteration, allowing you to process large datasets without loading everything into memory while maintaining responsiveness.
• Integration with async ecosystems: When combining Neo4j operations with httpx, aioredis, or other async libraries, a single event loop coordinates all I/O efficiently without thread-pool hopping.

Example: Asynchronous model definition and query

import asyncio
from neomodel.async_ import AsyncStructuredNode, StringProperty, IntegerProperty

class Person(AsyncStructuredNode):
    name = StringProperty()
    age = IntegerProperty()

async def main():
    # Await the save operation

    john = await Person(name="John", age=30).save()
    
    # Async iteration over results

    async for person in Person.nodes.filter(age__gt=20):
        print(person.name, person.age)

# Run the coroutine

asyncio.run(main())

Key Differences in Transaction Handling

Transaction syntax differs between the two APIs to accommodate their respective execution models and prevent blocking the event loop.

In the synchronous API, transactions use a standard context manager that blocks until the database commits or rolls back. The implementation in neomodel/sync_/transaction.py provides with db.transaction: syntax.

In the asynchronous API, transactions require an async context manager. The neomodel/async_/transaction.py file implements async with db.transaction: to properly yield control to the event loop during database operations.

Synchronous transaction example:

from neomodel import db, StructuredNode, StringProperty

class User(StructuredNode):
    username = StringProperty()

# Blocking transaction

with db.transaction:
    User(username="alice").save()
    User(username="bob").save()

Asynchronous transaction example:

from neomodel.async_ import db, AsyncStructuredNode, StringProperty

class User(AsyncStructuredNode):
    username = StringProperty()

async def create_users():
    async with db.transaction:
        await User(username="alice").save()
        await User(username="bob").save()

Migrating from Sync to Async Neomodel

Converting an existing neomodel codebase from synchronous to asynchronous requires systematic updates across four architectural layers.

1. Update imports – Replace from neomodel import StructuredNode with from neomodel.async_ import AsyncStructuredNode, and update database imports from neomodel to neomodel.async_.

2. Add await keywords – Every method that previously returned a value directly now returns a coroutine. Add await to save(), delete(), refresh(), cypher(), and get_or_create() calls.

3. Convert iteration – Change synchronous loops from for obj in Model.nodes.filter(...) to async for obj in Model.nodes.filter(...) to properly handle the AsyncNodeSet iterator defined in neomodel/async_/match.py.

4. Wrap entry points – Ensure your application entry points execute within an event loop using asyncio.run(), or use framework-specific patterns like FastAPI's automatic handling of async endpoints.

The neomodel project maintains deliberate feature parity between implementations, so any method available in StructuredNode has a corresponding AsyncStructuredNode equivalent with identical semantics.

Summary

• Choose the synchronous API when building traditional web applications with Flask or Django, writing simple scripts, or integrating with blocking third-party libraries like pandas.
• Choose the asynchronous API when working with FastAPI, aiohttp, or other async frameworks, or when handling high-concurrency workloads with hundreds of simultaneous database operations.
• Key implementation files include neomodel/sync_/node.py for StructuredNode and neomodel/async_/node.py for AsyncStructuredNode, with matching database wrappers in sync_/database.py and async_/database.py.
• Transaction syntax differs: use with db.transaction: for sync and async with db.transaction: for async operations.
• Migration path involves updating imports, adding await keywords, converting to async for iteration, and ensuring proper event loop execution.

Frequently Asked Questions

Can I mix sync and async neomodel code in the same application?

Mixing synchronous and asynchronous neomodel APIs in the same process is not recommended and can lead to deadlocks or runtime errors. The synchronous API blocks the thread until database operations complete, which freezes the event loop required by the async API. If you must use both, isolate them in separate processes or use asyncio.to_thread() to run sync code in a thread pool, though this adds significant complexity.

Does the async neomodel API support all features of the sync version?

Yes, the async neomodel API maintains complete feature parity with the synchronous version according to the source code architecture. Every method available on StructuredNode in neomodel/sync_/node.py has a corresponding async def implementation in neomodel/async_/node.py. This includes CRUD operations, query filtering via AsyncNodeSet, transaction management, and relationship handling.

How do I handle transactions in async neomodel compared to sync?

Transaction handling syntax reflects the underlying execution model. For synchronous code in neomodel/sync_/transaction.py, use with db.transaction: to create a blocking context manager. For asynchronous operations in neomodel/async_/transaction.py, use async with db.transaction: to properly yield control to the event loop during commit and rollback operations. Both approaches ensure atomicity, but the async version prevents blocking other coroutines during database waits.

What is the performance difference between sync and async neomodel?

For single requests, latency is nearly identical since both APIs use the same underlying Neo4j driver. The performance difference emerges under high concurrency: the synchronous API in neomodel/sync_/database.py blocks threads, limiting throughput to your thread pool size and increasing memory usage per connection. The asynchronous API in neomodel/async_/database.py multiplexes many concurrent queries over a single connection pool using coroutines, delivering significantly higher throughput with lower memory overhead when handling hundreds of simultaneous operations.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}