neo4j-driver-python-skill
Neo4j Python Driver v6 — driver lifecycle, execute_query, managed and explicit transactions, async (AsyncGraphDatabase), result handling, data type mapping, error handling, UNWIND batching, connection pool tuning, and causal consistency. Use when writing Python code that connects to Neo4j via GraphD
By neo4j-contrib · 647 installs
npx skills add neo4j-contrib/neo4j-skills --skill neo4j-driver-python-skill
Source repository · Upstream listing
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
Python =3.10 required for v6.x. Python 3.14 supported [6.1+]. Pandas 3 and PyArrow 23/24 supported [6.2+].
Environment Variables
Load connection config from environment — never hardcode credentials.
.env file format:
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.
URI schemes:
Scheme Use
neo4j+s:// TLS + cluster routing — Aura default
neo4j:// Unencrypted + cluster routing
bolt+s:// TLS, single instance
bolt:// Unencrypted, single instance
Auth options: ("user", "pass") tuple, basic auth() , bearer auth("jwt") , kerberos auth("b64") .
Choosing the Right API
API Use when Auto retry Streaming
driver.execute query() Most queries — simple, safe default ✅ ❌ eager
session.execute read/write() Large results / multiple queries in one tx ✅ ✅
session.run() LOAD CSV , CALL {} IN TRANSACTIONS , scripts ⚠️ one shot [6.2+] ✅
AsyncGraphDatabase asyncio applications ✅ ✅
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
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:
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.
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):
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.
Async API
Mirror of sync API — replace GraphDatabase with AsyncGraphDatabase , await every call.
FastAPI lifespan pattern:
Parallel queries with asyncio.gather :
Never use sync GraphDatabase in asyncio — blocks the event loop.
Full async patterns → [references/async.md](references/async.md)
Error Handling
Catch ConstraintError before Neo4jError — it is a subclass and will be swallowed otherwise.
Result Access & Null Safety
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.
Node/Relationship/temporal access:
Full type mapping table → [references/data types.md](references/data types.md)
Batch Writes with UNWIND
Pass list[dict] — only shape the driver serializes correctly for UNWIND .
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:
Session exhaustion: each open session holds a connection. Always use with driver.session(...) as session .
Full performance patterns → [references/performance.md](references/performance.md)
Common Errors
Mistake Fix
f string / .format() Cypher params Use $param placeholders always
Param name ending with Pass via parameters ={"key ": val}
Omitting database Always set — saves a round trip every call
Returning Result from tx callback Consume to list inside callback
Side effects in execute read/write callback Move outside — callback may retry
Passing dataclass/Pydantic as param Convert to dict first
UNWIND with list of objects list[dict] only
record.get() for absent key detection "key" in record.keys() for absent; .get() returns None for both absent and graph null
No .consume() after session.run() Commit timing undefined; call .consume()
Sync driver inside asyncio Use AsyncGraphDatabase — sync blocks event loop
Async driver created per request Singleton — create once at startup
Leaked sessions with driver.session(...) as session always
json.dumps(record.data()) with node/temporal Project scalars in Cypher or convert explicitly
result["name"] on EagerResult Index result.records[0]["name"] or unpack records, , = ...
Result.single() returns None for 0 results It raises — use single(strict=False)
@unit of work on lambda Use named function
Neo4jError caught before ConstraintError Catch ConstraintError first — it's a subclass
neo4j driver package name Package is neo4j since v6; neo4j driver deprecated
References
Load on demand:
[references/async.md](references/async.md) — full async patterns: managed transactions, result methods, concurrency
[references/data types.md](references/data types.md) — complete Python↔Cypher type mapping, temporal conversion, graph object API, spatial types (CartesianPoint/WGS84Point)
[references/performance.md](references/performance.md) — connection pool, lazy streaming, threading vs asyncio, bookmarks/causal consistency
[references/transactions.md](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]