Graph Data Modeling

Async Neo4j driver patterns for concurrent loads

You are loading many small, independent batches into Neo4j — one per source file, per tenant, per partition — and each batch spends most of its wall-clock time waiting on network round-trips rather than on the database doing work. This is the shape of load where asyncio pays off: with the async driver, a single thread can have dozens of batches in flight at once, overlapping their I/O waits instead of serialising them. This task shows how to drive Neo4j 5.x with AsyncGraphDatabase, run an async load coroutine per batch, bound the fan-out with an asyncio.Semaphore so you never overwhelm the pool or the server, and — just as importantly — recognise when async is the wrong tool and server-side batching or plain sync loading wins. This task is part of Python Driver Integration Patterns.

Prerequisites

Async overlaps waiting, it does not add parallelism

The async driver runs on one thread and one event loop. Its win is that while one coroutine is awaiting a Bolt round-trip, the loop runs another coroutine’s work, so the round-trip latencies overlap instead of stacking. That helps precisely when the workload is I/O-bound: many independent batches, each modest in size, where the bottleneck is round-trip latency rather than server CPU or lock contention. It does not speed up a single large batch, and it does not add database throughput when the server is already the bottleneck — the event loop is still one thread issuing queries. The load must be bounded so the fan-out never exceeds what the connection pool and server can absorb; an unbounded asyncio.gather over ten thousand batches will simply exhaust the pool and time out.

Bounded async fan-out into one driver Left: a stack of many batch coroutines. Centre: an asyncio.Semaphore gate admitting a bounded number of permits. Admitted coroutines flow to a shared async driver and pool on the right, each awaiting a write to Neo4j; coroutines without a permit wait at the gate. batch coro batch coro batch coro batch coro waiting waiting Semaphore N permits permit permit permit admitted Async driver shared pool await execute_write → Neo4j
The semaphore caps in-flight coroutines so the shared pool is never oversubscribed.

Core implementation

The async API mirrors the sync one: AsyncGraphDatabase.driver, async with sessions, and await session.execute_write(fn). The load coroutine below wraps its work in a semaphore acquire so only N batches are ever in flight, then gather launches every batch and lets the loop overlap their I/O.

python
import asyncio
from neo4j import AsyncGraphDatabase, AsyncManagedTransaction

async def _upsert(tx: AsyncManagedTransaction, rows: list[dict]) -> int:
    # Same idempotent, side-effect-free MERGE as the sync case: managed retry
    # re-runs it on transient faults, so it must converge on replay.
    result = await tx.run(
        """
        UNWIND $rows AS row
        MERGE (n:Entity {id: row.id})
        ON CREATE SET n += row.properties
        ON MATCH  SET n += row.properties
        RETURN count(n) AS upserted
        """,
        rows=rows,
    )
    record = await result.single()
    return record["upserted"]


async def load_batch(driver, sem: asyncio.Semaphore, rows: list[dict]) -> int:
    # BOUND CONCURRENCY. The semaphore admits at most N coroutines into the
    # pool-consuming region at once, so in-flight sessions never exceed the
    # connection pool. Everything outside the `async with sem` is free to queue.
    async with sem:
        async with driver.session(database="neo4j") as session:
            # `await` yields the event loop during every round-trip, so other
            # admitted batches make progress while this one waits on Bolt.
            return await session.execute_write(_upsert, rows)


async def run(uri, auth, batches: list[list[dict]], *, concurrency: int = 16) -> int:
    # One shared async driver for the whole run; pool sized to `concurrency`.
    async with AsyncGraphDatabase.driver(
        uri, auth=auth, max_connection_pool_size=concurrency + 4
    ) as driver:
        await driver.verify_connectivity()
        sem = asyncio.Semaphore(concurrency)          # the fan-out bound
        # gather launches every batch; the semaphore keeps only `concurrency`
        # of them touching the pool at any instant — the rest await a permit.
        counts = await asyncio.gather(
            *(load_batch(driver, sem, b) for b in batches)
        )
        return sum(counts)


# asyncio.run(run(uri, auth, batches, concurrency=16))

Two invariants make this safe. First, the semaphore bound must be less than or equal to the pool size, or coroutines that clear the semaphore will still block waiting for a connection — moving the bottleneck without relieving it. Second, the driver is created once for the whole run and shared, exactly as in the sync model; a driver per batch defeats pooling just as badly in async code.

Validation & verification

Confirm that concurrency is actually bounded and that overlap is real, not just launched. After a run, the total should match the sum of batch sizes with no duplicates — proof the awaited execute_write stayed idempotent under overlap:

cypher
// No duplicates despite many overlapping coroutines: the constraint-backed
// MERGE converged regardless of interleaving.
MATCH (n:Entity)
RETURN count(n) AS nodes, count(DISTINCT n.id) AS distinct_ids;   // must be equal

To verify the semaphore is doing its job, instrument the coroutine to record how many are inside the async with sem block at once; the peak must never exceed concurrency:

python
# Drop-in check: peak in-flight must equal `concurrency`, never exceed it.
inflight = 0
peak = 0
async def load_batch_checked(driver, sem, rows):
    global inflight, peak
    async with sem:
        inflight += 1; peak = max(peak, inflight)
        try:
            async with driver.session(database="neo4j") as s:
                return await s.execute_write(_upsert, rows)
        finally:
            inflight -= 1
# after gather: assert peak <= concurrency

Edge cases & gotchas

1. Unbounded gather exhausts the pool. Calling asyncio.gather over every batch with no semaphore launches all of them at once; the surplus coroutines pile onto the pool and raise acquisition timeouts. Always gate with a semaphore whose bound is at or below the pool size.

2. Reaching for async when the server is the bottleneck. If a load is limited by database CPU, lock contention on shared nodes, or one very large batch, async adds coordination overhead without adding throughput — the single event loop still issues one query at a time. For a single huge offline load, server-side CALL { … } IN TRANSACTIONS batches inside the database and applies native backpressure; for CPU-bound transforms, plain sync batching across a process pool is simpler. Reserve async for many small, independent, I/O-bound batches.

3. Blocking the event loop. A synchronous call inside a coroutine — a blocking file read, a time.sleep, a CPU-heavy transform — stalls the whole loop and erases the overlap async exists to provide. Keep coroutines await-only, and push blocking work to asyncio.to_thread or a separate stage.

Related pages