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. Python 3.14 supported [6.1+]. Pandas 3 and PyArrow 23/24 supported [6.2+]. PyArrow 25 and Bolt 6.1 uuid.UUID values supported [6.3+]; driver-created SSL contexts honour SSLKEYLOGFILE [6.3+].

Neo4j 2026.08+ UUID properties require neo4j>=6.3 to round-trip as uuid.UUID.

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"),