Neo4j Driver Python Skill

When to Use

• Writing Python code that connects to Neo4j
• Setting up driver, sessions, transactions, or async patterns
• Debugging result handling, serialization, or UNWIND batching
• Reviewing Neo4j driver usage in Python code

When NOT to Use

• Writing/optimizing Cypher → neo4j-cypher-skill
• Driver version upgrades → neo4j-migration-skill
• GraphRAG pipelines (neo4j-graphrag package) → neo4j-graphrag-skill

---

Installation

pip install neo4j                  # package name is `neo4j`, NOT `neo4j-driver` (deprecated since v6)
pip install neo4j-rust-ext         # optional: 3–10× faster serialization, same API

Python >=3.10 required for v6.x.

---

Environment Variables

Load connection config from environment — never hardcode credentials.

import os
from dotenv import load_dotenv   # pip install python-dotenv

load_dotenv(".env")   # reads NEO4J_URI / NEO4J_USERNAME / NEO4J_PASSWORD / NEO4J_DATABASE

URI      = os.getenv("NEO4J_URI",      "neo4j://localhost:7687")
USER     = os.getenv("NEO4J_USERNAME", "neo4j")
PASSWORD = os.getenv("NEO4J_PASSWORD", "")
DATABASE = os.getenv("NEO4J_DATABASE", "neo4j")

.env file format:

NEO4J_URI=neo4j+s://xxx.databases.neo4j.io
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=secret
NEO4J_DATABASE=neo4j

Add .env to .gitignore. Without python-dotenv, use export in shell or os.getenv directly.

---

Driver Lifecycle

Create one Driver per application. Thread-safe, expensive to create. Never create per-request.

from neo4j import GraphDatabase

URI  = "neo4j+s://xxx.databases.neo4j.io"   # Aura
AUTH = ("neo4j", "password")

# Context manager — preferred for scripts
with GraphDatabase.driver(URI, auth=AUTH) as driver:
    driver.verify_connectivity()
    # ... work ...

# Long-lived singleton (service / web app)
driver = GraphDatabase.driver(URI, auth=AUTH)
driver.verify_connectivity()
# on shutdown:
driver.close()

URI schemes:

Auth options: ("user", "pass") tuple, basic_auth(), bearer_auth("jwt"), kerberos_auth("b64").

---

Choosing the Right API

session.run() retry [6.2+]: single immediate retry on DBMS-marked idempotent errors only (currently admission control). Disable with disable_auto_commit_retries=True at driver or session level.

---

execute_query — Default API

from neo4j import GraphDatabase, RoutingControl

# Tuple unpacking — most common
records, summary, keys = driver.execute_query(
    "MATCH (p:Person {name: $name})-[:KNOWS]->(f) RETURN f.name AS name",
    name="Alice",
    routing_=RoutingControl.READ,   # route reads to replicas
    database_="neo4j",              # always specify — saves a round-trip
)
for record in records:
    print(record["name"])
print(summary.result_available_after, "ms")

# Write — check counters
summary = driver.execute_query(
    "CREATE (p:Person {name: $name, age: $age})",
    name="Bob", age=30,
    database_="neo4j",
).summary
print(summary.counters.nodes_created)

Trailing-underscore convention — config kwargs end with _ (database_, routing_, auth_, result_transformer_, bookmark_manager_). No query parameter name may end with _; pass those via parameters_={"key_": val}.

Never f-string or format Cypher. Always $param — prevents injection and enables plan caching.

result_transformer_ — reshape before return:

import neo4j
df      = driver.execute_query("MATCH (p:Person) RETURN p.name, p.age", database_="neo4j",
                                result_transformer_=neo4j.Result.to_df)
record  = driver.execute_query("MATCH (p:Person {name:$n}) RETURN p", n="Alice", database_="neo4j",
                                result_transformer_=neo4j.Result.single)   # raises if 0 or 2+ results

Result.single() raises ResultNotSingleError on zero results (not just 2+). Use single(strict=False) for None-on-empty.

---

Managed Transactions (execute_read / execute_write)

Use for large results or multiple queries in one transaction.

with driver.session(database="neo4j") as session:

    def get_people(tx):
        result = tx.run("MATCH (p:Person) WHERE p.name STARTS WITH $pfx RETURN p.name AS name",
                        pfx="Al")
        return [r["name"] for r in result]   # consume INSIDE callback — Result invalid after tx closes

    names = session.execute_read(get_people)

    def create_person(tx):
        tx.run("CREATE (p:Person {name: $name})", name="Carol")

    session.execute_write(create_person)

Result lifetime — Result is a lazy cursor backed by the open transaction. Returning it unconsumed raises ResultConsumedError. Always collect to list inside the callback.

Callback may retry on transient failures — keep callbacks idempotent; move side effects (HTTP calls, emails) outside the callback.

Timeout/metadata via @unit_of_work (named functions only — cannot decorate lambdas):

from neo4j import unit_of_work

@unit_of_work(timeout=5.0, metadata={"app": "svc", "user": user_id})
def get_people(tx):
    return [r["name"] for r in tx.run("MATCH (p:Person) RETURN p.name AS name")]

session.execute_read(get_people)

---

Implicit Transactions (session.run)

Use only for LOAD CSV, CALL {} IN TRANSACTIONS, or quick scripts. session.run() does a single immediate retry on idempotent (DBMS-marked) errors only [6.2+]; other errors do not retry.

with driver.session(database="neo4j") as session:
    result = session.run("CREATE (p:Person {name: $name})", name="Alice")
    summary = result.consume()   # call consume() to guarantee commit before proceeding
    print(summary.counters.nodes_created)

# Opt out of one-shot retry [6.2+] — driver- or session-level
driver = GraphDatabase.driver(URI, auth=AUTH, disable_auto_commit_retries=True)
with driver.session(database="neo4j", disable_auto_commit_retries=True) as session:
    session.run("...")

---

Async API

Mirror of sync API — replace GraphDatabase with AsyncGraphDatabase, await every call.

from neo4j import AsyncGraphDatabase
import asyncio

# Singleton — same rule as sync: never create per-request
driver = AsyncGraphDatabase.driver(URI, auth=AUTH)

async def main():
    records, _, _ = await driver.execute_query(
        "MATCH (p:Person) RETURN p.name AS name",
        database_="neo4j", routing_=RoutingControl.READ,
    )
    print([r["name"] for r in records])
    await driver.close()

asyncio.run(main())

FastAPI lifespan pattern:

from contextlib import asynccontextmanager
from fastapi import FastAPI

_driver = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global _driver
    _driver = AsyncGraphDatabase.driver(URI, auth=AUTH)
    await _driver.verify_connectivity()
    yield
    await _driver.close()

app = FastAPI(lifespan=lifespan)

Parallel queries with asyncio.gather:

results = await asyncio.gather(
    driver.execute_query("MATCH (a:Artist) RETURN a.name AS name", database_="neo4j"),
    driver.execute_query("MATCH (v:Venue)  RETURN v.name AS name",  database_="neo4j"),
)

Never use sync `GraphDatabase` in asyncio — blocks the event loop.

Full async patterns → references/async.md

---

Error Handling

from neo4j.exceptions import (
    Neo4jError, ServiceUnavailable, TransientError,
    AuthError, ConstraintError,
)

try:
    driver.execute_query("...", database_="neo4j")
except AuthError:
    ...  # bad credentials
except ServiceUnavailable:
    ...  # no servers reachable
except ConstraintError as e:
    # unique/existence constraint violation — catch BEFORE Neo4jError (it's a subclass)
    print(e.code, e.message)
except TransientError as e:
    # raised only after retries exhausted (execute_query retries automatically)
    print(e.code)
except Neo4jError as e:
    print(e.code, e.message, e.gql_status)

Catch ConstraintError before Neo4jError — it is a subclass and will be swallowed otherwise.

---

Result Access & Null Safety

record = records[0]
record["name"]               # by key — KeyError if absent
record[0]                    # by index
record.get("name")           # None for absent key OR graph null
record.get("name", "Unknown")
d = record.data()            # dict — values still driver objects for Node/Rel/temporal types

record.data() is not JSON-safe if result contains Node, Relationship, Path, or neo4j.time.* values. Project scalar fields in Cypher instead of returning whole nodes.

# ❌ raises TypeError on json.dumps
records, _, _ = driver.execute_query("MATCH (p:Person) RETURN p", database_="neo4j")
json.dumps(records[0].data())

# ✅ project scalars
records, _, _ = driver.execute_query(
    "MATCH (p:Person) RETURN p.name AS name, p.age AS age", database_="neo4j")
json.dumps(records[0].data())   # safe

Node/Relationship/temporal access:

node = record["p"]           # neo4j.graph.Node
node.element_id              # stable within this transaction only
node.labels                  # frozenset({'Person'})
dict(node)                   # all properties as plain dict

rel  = record["r"]           # neo4j.graph.Relationship
rel.type                     # 'KNOWS'

dt = record["created_at"]    # neo4j.time.DateTime
dt.to_native()               # datetime.datetime (loses sub-µs precision)

Full type mapping table → references/data-types.md

---

Batch Writes with UNWIND

Pass list[dict] — only shape the driver serializes correctly for UNWIND.

people = [{"name": "Alice", "age": 30}, {"name": "Bob", "age": 25}]
driver.execute_query(
    "UNWIND $rows AS row MERGE (p:Person {name: row.name}) SET p.age = row.age",
    rows=people,
    database_="neo4j",
)

Custom objects and dataclasses must be converted to dict before passing as parameters.

---

Performance

• Always set database_ / database= — omitting triggers a home-database round-trip per call.
• execute_read routes to replicas automatically; use routing_=RoutingControl.READ with execute_query.
• Batch writes: one execute_write callback for the whole list > one tx per item.
• Large results: stream lazily inside execute_read callback; execute_query is always eager.

Connection pool tuning:

driver = GraphDatabase.driver(URI, auth=AUTH,
    max_connection_pool_size=50,        # default 100
    connection_acquisition_timeout=30,  # seconds to wait for free connection
    max_connection_lifetime=3600,       # seconds; recycles stale connections
    connection_timeout=15,
    keep_alive=True,
)

Session exhaustion: each open session holds a connection. Always use with driver.session(...) as session.

Full performance patterns → references/performance.md

---

Common Errors

---

References

Load on demand:

• references/async.md — full async patterns: managed transactions, result methods, concurrency
• references/data-types.md — complete Python↔Cypher type mapping, temporal conversion, graph object API, spatial types (CartesianPoint/WGS84Point)
• references/performance.md — connection pool, lazy streaming, threading vs asyncio, bookmarks/causal consistency
• references/transactions.md — explicit transactions, rollback, commit uncertainty, unit_of_work details

Docs:

• https://neo4j.com/docs/python-manual/current/
• https://neo4j.com/docs/api/python-driver/current/

---

Checklist

• [ ] Package installed as neo4j (not neo4j-driver)
• [ ] One Driver instance created at startup; shared everywhere
• [ ] verify_connectivity() called at startup
• [ ] database_ / database= set on every call
• [ ] $param placeholders used — no f-strings or .format()
• [ ] Result consumed inside tx callback (not returned raw)
• [ ] Sessions used as context managers (with driver.session(...) as session)
• [ ] ConstraintError caught before Neo4jError
• [ ] AsyncGraphDatabase used in asyncio code (not sync driver)
• [ ] Async driver created once at app startup (not per request)
• [ ] Side effects outside execute_read/write callbacks
• [ ] UNWIND batches use list[dict]

neo4j-driver-python-skill

Skill for writing Python applications that connect to Neo4j using the official Neo4j Python Driver.

Covers:

• Installation and driver lifecycle (singleton pattern, verify_connectivity)
• URI schemes and auth options (Aura, bolt, bearer, Kerberos)
• execute_query — default API with RoutingControl, result_transformer_, trailing-underscore convention
• Managed transactions (execute_read / execute_write) — retry safety, result lifetime, @unit_of_work
• Implicit transactions (session.run) — LOAD CSV, CALL {} IN TRANSACTIONS
• Async driver (AsyncGraphDatabase) — FastAPI lifespan pattern, asyncio.gather
• Error handling — ConstraintError, ServiceUnavailable, TransientError, GQL status codes
• Result access — Record, record.data(), JSON serialization gotchas
• Data type mapping — Python ↔ Cypher, temporal types, graph objects (Node, Relationship)
• UNWIND batch writes (list[dict] only)
• Connection pool tuning and session exhaustion
• Causal consistency and bookmarks

Version / compatibility:

• Driver v6.x (Jan 2026+) — package name is neo4j, not neo4j-driver
• Python ≥ 3.10 required

Not covered:

• Cypher query authoring → use neo4j-cypher-skill
• Driver version upgrades / breaking changes → use neo4j-migration-skill
• GraphRAG pipelines (neo4j-graphrag package) → use neo4j-graphrag-skill

Install:

pip install neo4j

Async Driver — Full Reference

Setup

from neo4j import AsyncGraphDatabase, RoutingControl
import asyncio

URI  = "neo4j+s://xxx.databases.neo4j.io"
AUTH = ("neo4j", "password")

# Singleton — never create per-request
driver = AsyncGraphDatabase.driver(URI, auth=AUTH)
await driver.verify_connectivity()
# on shutdown:
await driver.close()

Async Managed Transactions

async def get_people(tx):
    result = await tx.run("MATCH (p:Person) RETURN p.name AS name")
    return await result.values()   # consume INSIDE callback

async def create_person(tx, name: str):
    await tx.run("MERGE (p:Person {name: $name})", name=name)

async def run_queries(driver):
    async with driver.session(database="neo4j") as session:
        people = await session.execute_read(get_people)
        await session.execute_write(create_person, "Carol")

Async Result Methods

FastAPI Lifespan Pattern

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from neo4j import AsyncGraphDatabase, RoutingControl

_driver = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global _driver
    _driver = AsyncGraphDatabase.driver(URI, auth=AUTH)
    await _driver.verify_connectivity()
    yield
    await _driver.close()

app = FastAPI(lifespan=lifespan)

def get_driver():
    return _driver

@app.get("/people")
async def get_people(driver=Depends(get_driver)):
    records, _, _ = await driver.execute_query(
        "MATCH (p:Person) RETURN p.name AS name",
        database_="neo4j",
        routing_=RoutingControl.READ,
    )
    return [r["name"] for r in records]

Concurrency with asyncio.gather

async def run_concurrent(driver):
    results = await asyncio.gather(
        driver.execute_query("MATCH (a:Artist) RETURN a.name AS name", database_="neo4j"),
        driver.execute_query("MATCH (v:Venue)  RETURN v.name AS name",  database_="neo4j"),
    )
    artists = [r["name"] for r in results[0].records]
    venues  = [r["name"] for r in results[1].records]

Common Async Mistakes

# ❌ Sync driver in asyncio — blocks event loop
async def bad():
    with GraphDatabase.driver(URI, auth=AUTH) as driver:
        records, _, _ = driver.execute_query("MATCH (p:Person) RETURN p")

# ✅ Async driver
async def good():
    async with AsyncGraphDatabase.driver(URI, auth=AUTH) as driver:
        records, _, _ = await driver.execute_query("MATCH (p:Person) RETURN p")

# ❌ Async driver created per request — rebuilds connection pool every time
async def handle_request(name: str):
    async with AsyncGraphDatabase.driver(URI, auth=AUTH) as driver:
        records, _, _ = await driver.execute_query("...", database_="neo4j")

# ✅ Singleton at startup
_driver = AsyncGraphDatabase.driver(URI, auth=AUTH)

async def handle_request(name: str):
    records, _, _ = await _driver.execute_query("...", database_="neo4j")

Data Types — Python ↔ Cypher Mapping

Parameter Types (allowed)

Custom classes, dataclasses, Pydantic models, and enums are not auto-serialized — convert to dict or primitives first.

from dataclasses import dataclass, asdict

@dataclass
class Person:
    name: str
    age: int

p = Person("Alice", 30)

# ❌ Fails
driver.execute_query("CREATE (p:Person $props)", props=p, database_="neo4j")

# ✅ Convert to dict
driver.execute_query("CREATE (p:Person $props)", props=asdict(p), database_="neo4j")

Graph Object API

# Node — neo4j.graph.Node
node = record["p"]
node.element_id        # stable within this transaction; don't use across transactions
node.labels            # frozenset({'Person'})
node["name"]           # property access by key
dict(node)             # all properties as plain dict

# Relationship — neo4j.graph.Relationship
rel = record["r"]
rel.type               # 'KNOWS'
rel.start_node.element_id
rel.end_node.element_id
rel["since"]           # property
dict(rel)              # all properties as plain dict

Temporal Types

from neo4j.time import DateTime, Date, Time, Duration

dt = record["created_at"]   # neo4j.time.DateTime
dt.to_native()              # datetime.datetime — loses sub-microsecond precision
str(dt)                     # ISO 8601 string — JSON-safe

# Pass Python datetime as a parameter — driver converts automatically
from datetime import datetime, timezone
driver.execute_query("CREATE (e:Event {at: $ts})", ts=datetime.now(timezone.utc), database_="neo4j")

# Duration — access .days / .months (not .inDays / .inMonths)
dur = record["tenure"]      # neo4j.time.Duration
dur.days
dur.months

JSON Serialization

record.data() returns a dict but Node, Relationship, Path, and neo4j.time.* values are still driver objects — not JSON-safe.

# ❌ Raises TypeError if result contains node/rel/temporal
json.dumps(records[0].data())

# ✅ Project scalars in Cypher
records, _, _ = driver.execute_query(
    "MATCH (p:Person) RETURN p.name AS name, p.age AS age, toString(p.created_at) AS created_at",
    database_="neo4j",
)
json.dumps(records[0].data())   # safe — all scalars

# ✅ Extract node properties manually
node = records[0]["p"]
props = dict(node)              # plain dict — json-safe if all property types are primitives

Spatial Types

from neo4j.spatial import CartesianPoint, WGS84Point

# 2D Cartesian (SRID 7203)
pt2d = CartesianPoint((1.23, 4.56))
print(pt2d.x, pt2d.y, pt2d.srid)   # 1.23, 4.56, 7203

# 3D Cartesian (SRID 9157)
pt3d = CartesianPoint((1.23, 4.56, 7.89))
x, y, z = pt3d   # destructuring

# 2D WGS-84 (SRID 4326)
ldn = WGS84Point((-0.118092, 51.509865))
print(ldn.longitude, ldn.latitude, ldn.srid)   # -0.118092, 51.509865, 4326

# 3D WGS-84 (SRID 4979)
shard = WGS84Point((-0.086500, 51.504501, 310))
longitude, latitude, height = shard

# Distance (same SRID only — returns None if SRIDs differ)
records, _, _ = driver.execute_query(
    "RETURN point.distance($p1, $p2) AS distance",
    p1=CartesianPoint((1, 1)), p2=CartesianPoint((10, 10)),
    database_="neo4j",
)
distance = records[0]["distance"]   # float64

Pass points as parameters — serialized automatically. Read back via destructuring or .x/.y/.z.

Null Safety

.get() cannot distinguish absent key from graph null — use "key" in record.keys() when the distinction matters.

# Optional column from OPTIONAL MATCH
if "city" in record.keys() and record["city"] is not None:
    city = record["city"]
else:
    city = "Unknown"

Performance & Scalability

Connection Pool Configuration

driver = GraphDatabase.driver(URI, auth=AUTH,
    max_connection_pool_size=50,        # default 100; tune to workload
    connection_acquisition_timeout=30,  # seconds to wait for free connection
    max_connection_lifetime=3600,       # seconds; recycles stale connections
    connection_timeout=15,              # seconds to establish new connection
    keep_alive=True,                    # TCP keepalive
)

Each open session holds a connection. Leaked sessions exhaust the pool — new sessions block until connection_acquisition_timeout then raise ClientError. Always use with driver.session(...) as session.

Batch Writes — Three Patterns

UNWIND (best for bulk import)

rows = [{"name": "Alice", "age": 30}, {"name": "Bob", "age": 25}]
driver.execute_query(
    "UNWIND $rows AS row MERGE (p:Person {name: row.name}) SET p.age = row.age",
    rows=rows, database_="neo4j",
)

Group in one managed transaction

# ❌ One tx per item — high overhead
for item in items:
    driver.execute_query("CREATE (n:Node {id: $id})", id=item["id"], database_="neo4j")

# ✅ One callback for the whole batch
def bulk_create(tx):
    for item in items:
        tx.run("CREATE (n:Node {id: $id})", id=item["id"])

with driver.session(database="neo4j") as session:
    session.execute_write(bulk_create)

CALL IN TRANSACTIONS (very large data — use via session.run, not execute_query)

UNWIND $rows AS row
CALL (row) {
  MERGE (p:Person {name: row.name})
} IN TRANSACTIONS OF 1000 ROWS ON ERROR CONTINUE

Lazy vs Eager Loading

# execute_query is always eager — fine for small/medium results
records, _, _ = driver.execute_query("MATCH (p:Person) RETURN p", database_="neo4j")

# Large results — stream lazily inside managed transaction
def process_large_result(tx):
    result = tx.run("MATCH (p:Person) RETURN p.name AS name")
    for record in result:           # one record at a time
        process(record["name"])     # don't build a list

with driver.session(database="neo4j") as session:
    session.execute_read(process_large_result)

Threading vs asyncio

The Python GIL limits CPU parallelism for threads; both threads and asyncio overlap on I/O.

# Sync threading — OK for moderate I/O concurrency
from concurrent.futures import ThreadPoolExecutor

def query(name):
    records, _, _ = driver.execute_query(
        "MATCH (p:Person {name: $name}) RETURN p", name=name, database_="neo4j"
    )
    return records

with ThreadPoolExecutor(max_workers=10) as pool:
    results = list(pool.map(query, names))

# asyncio — preferred for high-concurrency workloads
async def run_all(names):
    tasks = [
        driver.execute_query("MATCH (p:Person {name: $name}) RETURN p",
                              name=name, database_="neo4j")
        for name in names
    ]
    return await asyncio.gather(*tasks)

Causal Consistency & Bookmarks

Within a single session, queries are automatically causally chained. Across sessions — use execute_query (shares BookmarkManager automatically), or pass bookmarks explicitly:

from neo4j import Bookmarks

with driver.session(database="neo4j") as session_a:
    session_a.execute_write(lambda tx: tx.run("MERGE (p:Person {name: 'Alice'})"))
    bookmarks_a = session_a.last_bookmarks()

with driver.session(database="neo4j") as session_b:
    session_b.execute_write(lambda tx: tx.run("MERGE (p:Person {name: 'Bob'})"))
    bookmarks_b = session_b.last_bookmarks()

combined = Bookmarks.from_raw_values(
    *bookmarks_a.raw_values, *bookmarks_b.raw_values
)

with driver.session(database="neo4j", bookmarks=combined) as session_c:
    session_c.execute_write(
        lambda tx: tx.run("MATCH (a:Person {name:'Alice'}), (b:Person {name:'Bob'}) "
                          "MERGE (a)-[:KNOWS]->(b)")
    )

execute_query shares a BookmarkManager automatically — usually sufficient.

Transactions — Full Reference

Explicit Transactions

Use when a transaction spans multiple functions or coordinates with external state.

with driver.session(database="neo4j") as session:
    tx = session.begin_transaction()
    try:
        do_part_a(tx)
        do_part_b(tx)
        tx.commit()
    except Exception as e:
        tx.rollback()
        raise

def do_part_a(tx):
    tx.run("CREATE (p:Person {name: $name})", name="Alice")

Rollback Can Raise

tx.rollback() is a network call — if the connection is broken, it raises. Don't let it swallow the original exception:

try:
    tx.commit()
except Exception as original:
    try:
        tx.rollback()
    except Exception as rollback_err:
        original.__suppress_context__ = False
        raise rollback_err from original   # chain both exceptions
    raise

Commit Uncertainty

If tx.commit() raises a network-level exception, the commit may or may not have succeeded. Design writes as idempotent with MERGE and unique constraints so retrying is safe.

@unit_of_work — Timeout & Metadata

Attaches timeout and server metadata to a managed transaction callback.

from neo4j import unit_of_work

@unit_of_work(timeout=5.0, metadata={"app": "myService", "user": user_id})
def get_people(tx):
    return [r["name"] for r in tx.run("MATCH (p:Person) RETURN p.name AS name")]

session.execute_read(get_people)

Metadata appears in SHOW TRANSACTIONS and server query logs.

Cannot Decorate Lambdas

# ❌ Syntax error — cannot decorate a lambda inline
session.execute_write(
    @unit_of_work(timeout=5.0)
    lambda tx: tx.run("MERGE (p:Person {name: $name})", name="Alice")
)

# ❌ Also wrong — the original lambda is used, not the wrapped version
fn = lambda tx: tx.run("MERGE (p:Person {name: $name})", name="Alice")
unit_of_work(timeout=5.0)(fn)   # wraps fn, but not reassigned
session.execute_write(fn)       # uses original

# ✅ Named function with decorator
@unit_of_work(timeout=5.0, metadata={"app": "myService"})
def create_person(tx):
    tx.run("MERGE (p:Person {name: $name})", name="Alice")

session.execute_write(create_person)

# ✅ Assign the wrapped lambda explicitly
create_person = unit_of_work(timeout=5.0)(lambda tx: tx.run(
    "MERGE (p:Person {name: $name})", name="Alice"
))
session.execute_write(create_person)

Use named functions when timeout or metadata is needed; lambdas are fine for fire-and-forget callbacks.

Multiple tx.run() Calls

Calling tx.run() again before the first Result is consumed causes the driver to buffer the first result in memory. Safe, but can pull large results into RAM unexpectedly.

def multi_query_tx(tx):
    people = [r["name"] for r in tx.run("MATCH (p:Person) RETURN p.name AS name")]
    # first result consumed — safe to issue second query
    for name in people:
        tx.run("MERGE (:Person {name: $name})-[:VISITED]->(:City {name: 'London'})", name=name)
    return len(people)

Retry Safety

execute_read / execute_write callbacks may execute more than once on transient failures — keep them side-effect-free.

# ❌ Side effect fires on every retry
def dangerous_tx(tx):
    requests.post("https://api.example.com/notify")   # fires on every retry
    tx.run("CREATE (p:Person {name: $name})", name="Alice")

# ✅ Database work only; HTTP call made after confirmed success
def safe_tx(tx):
    tx.run("MERGE (p:Person {name: $name})", name="Alice")   # idempotent

session.execute_write(safe_tx)
requests.post("https://api.example.com/notify")   # outside callback

Repository Pattern

from neo4j import Driver, RoutingControl
from dataclasses import dataclass

@dataclass
class Person:
    name: str
    age: int

class PersonRepository:
    def __init__(self, driver: Driver, database: str = "neo4j"):
        self._driver = driver
        self._db = database

    def find_by_name_prefix(self, prefix: str) -> list[Person]:
        records, _, _ = self._driver.execute_query(
            "MATCH (p:Person) WHERE p.name STARTS WITH $prefix RETURN p.name AS name, p.age AS age",
            prefix=prefix,
            routing_=RoutingControl.READ,
            database_=self._db,
        )
        return [Person(name=r["name"], age=r["age"]) for r in records]

    def create(self, person: Person) -> None:
        self._driver.execute_query(
            "CREATE (p:Person {name: $name, age: $age})",
            name=person.name, age=person.age,
            database_=self._db,
        )

    def bulk_create(self, people: list[Person]) -> None:
        rows = [{"name": p.name, "age": p.age} for p in people]
        self._driver.execute_query(
            "UNWIND $rows AS row MERGE (p:Person {name: row.name}) SET p.age = row.age",
            rows=rows,
            database_=self._db,
        )