neo4j-driver-python-skill

适用场景

• 编写连接 Neo4j 的 Python 代码
• 设置驱动程序、会话、事务或异步模式
• 调试结果处理、序列化或 UNWIND 批处理
• 检查 Python 代码中的 Neo4j 驱动程序用法

不适用场景

• 编写/优化 Cypher → neo4j-cypher-skill
• 升级驱动程序版本 → neo4j-migration-skill
• GraphRAG 管道（neo4j-graphrag package）→ neo4j-graphrag-skill

---

安装

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

v6.x 要求 Python >=3.10。支持 Python 3.14 [6.1+]。支持 Pandas 3 和 PyArrow 23/24 [6.2+]。支持 PyArrow 25 和 Bolt 6.1 的 uuid.UUID 值 [6.3+]；驱动程序创建的 SSL 上下文会遵循 SSLKEYLOGFILE [6.3+]。支持 Python 3.15 [6.4+]。固定使用 neo4j>=6.4.1 — 6.4.0 在获取连接池连接时可能无限等待，并且会在 HTTP Query API 中对 bearer token 进行双重编码。

Neo4j 2026.08+ 的 UUID 属性需要 neo4j>=6.3 才能往返转换为 uuid.UUID。存储 UUID 需要 block store 格式（Enterprise）；Community（aligned format）会失败：storing properties of type UUID is not supported in aligned store format — 请存储 str(uuid)。

---

环境变量

从环境变量加载连接配置 — 切勿将凭据硬编码。

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 文件格式：

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

将 .env 添加到 .gitignore。如果没有 python-dotenv，请在 shell 中使用 export，或直接使用 os.getenv。

---

驱动程序生命周期

每个应用程序创建一个 Driver。它是线程安全的，且创建成本高。切勿为每个请求创建。

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 方案：

方案	用途
neo4j+s://	TLS + 集群路由 — Aura 默认值
neo4j://	未加密 + 集群路由
bolt+s://	TLS，单实例
bolt://	未加密，单实例
https:// / http://（例如 https://host:7474/）	使用 HTTP Query API 代替 Bolt [6.4+，预览版] — pip install "neo4j[http]"

身份验证选项：("user", "pass") 元组、basic_auth()、bearer_auth("jwt")、kerberos_auth("b64")。

HTTP Query API [6.4+，预览版 — 不保证 semver 兼容性]：每次调用都传入 database_ / database=（不解析 home database）；仅支持 basic（无 realm）或 bearer 身份验证。不支持或会忽略：事务元数据和超时、fetch_size（无背压）、通知筛选器；无法交换 VECTOR、UUID、UnsupportedType。如需使用这些功能 → 使用 neo4j+s:// / bolt+s://。

---

选择合适的 API

API	适用场景	自动重试	流式处理
driver.execute_query()	大多数查询 — 简单、安全的默认选择	✅	❌ 立即获取
session.execute_read/write()	大型结果集 / 在一个事务中执行多个查询	✅	✅
session.run()	LOAD CSV、CALL {} IN TRANSACTIONS、脚本	⚠️ 单次 [6.2+]	✅
AsyncGraphDatabase	asyncio 应用程序	✅	✅

session.run() 重试 [6.2+]：仅对 DBMS 标记为幂等的错误立即重试一次（目前为准入控制）。可在驱动程序或会话级别通过 disable_auto_commit_retries=True 禁用。

---

execute_query — 默认 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)

结尾下划线约定 — 配置关键字参数以 _ 结尾（database_、routing_、auth_、result_transformer_、bookmark_manager_）。查询参数名称不能以 _ 结尾；这类参数请通过 parameters_={"key_": val} 传递。

切勿对 Cypher 使用 f-string 或 format。 始终使用 $param — 防止注入并启用计划缓存。

result_transformer_ — 返回前重塑结果：

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)   # None if 0 rows; first record + warning if 2+

Result.single() 默认使用 strict=False：0 行 → None；2+ 行 → 第一条记录 + 警告（不抛出异常）。只有 single(strict=True) 会引发 ResultNotSingleError（0 行或 2+ 行）。当要求恰好返回一行时，请使用 strict=True，例如 result_transformer_=lambda r: r.single(strict=True)；否则请检查结果是否为 None。

---

托管事务（execute_read / execute_write）

用于处理大型结果集，或在一个事务中执行多个查询。

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 是由打开的事务支持的惰性游标。未消费就返回它会引发 ResultConsumedError。务必在回调中收集到 list。

回调可能会重试，以应对瞬时故障 — 确保回调具备幂等性；将副作用（HTTP 调用、邮件）移到回调之外。

通过 @unit_of_work 设置超时/元数据（仅限命名函数 — 不能装饰 lambda）：

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)

---

隐式事务（session.run）

仅用于 LOAD CSV、CALL {} IN TRANSACTIONS 或快速脚本。session.run() 仅会针对幂等（由 DBMS 标记）的错误立即重试一次 [6.2+]；其他错误不会重试。

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("...")

---

异步 API

与同步 API 相对应 — 将 GraphDatabase 替换为 AsyncGraphDatabase，并对每个调用使用 await。

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 生命周期模式：

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)

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

切勿在 asyncio 中使用同步的 GraphDatabase — 这会阻塞事件循环。

完整的异步模式 → references/async.md

---

错误处理

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)

先捕获 ConstraintError，再捕获 Neo4jError — 前者是后者的子类，否则会被后者一并捕获。

---

结果访问与空值安全

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 — Node → dict of properties, Relationship → tuple, Path → list; temporal values stay driver objects

record.data()：Node → 属性字典；Relationship → (start_props, type, end_props) 元组（自身属性会被丢弃）；Path → 列表。json.dumps 可以接受这些类型，但会丢失标签、元素 ID 和关系属性。neo4j.time.Date / Time / DateTime 仍是驱动程序对象 → json.dumps 会引发 TypeError。请在 Cypher 中提取所需的标量值（对时间类型使用 toString()）；不要返回完整实体。

# ❌ raises TypeError on json.dumps (temporal value)
records, _, _ = driver.execute_query("MATCH (p:Person) RETURN p.name AS name, p.created_at AS created_at", 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, toString(p.created_at) AS created_at", database_="neo4j")
json.dumps(records[0].data())   # safe

Node / Relationship / 时间类型访问：

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)

完整的类型映射表 → references/data-types.md

---

使用 UNWIND 批量写入

传入 list[dict] — 这是驱动程序能够正确序列化、供 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",
)

自定义对象和数据类必须先转换为 dict，再作为参数传入。

---

性能

• 始终设置 database_ / database= — 省略时，每次调用都会触发一次往返以查询主数据库。
• execute_read 会自动将请求路由到副本；使用 execute_query 时，请设置 routing_=RoutingControl.READ。
• 批量写入：整个列表使用一个 execute_write 回调 > 每个项目使用一个事务。
• 大型结果集：在 execute_read 回调中惰性流式读取；execute_query 始终会立即获取结果。

连接池调优：

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

会话耗尽：每个打开的会话都会占用一个连接。务必使用 with driver.session(...) as session。

完整的性能模式 → references/performance.md

---

常见错误

错误	修正方法
使用 f-string / .format() 拼接 Cypher 参数	始终使用 $param 占位符
参数名以 _ 结尾	通过 parameters_={"key_": val} 传递
忽略 database_	始终设置它——每次调用都能省去一次往返
从事务回调返回 Result	在回调中将其读取到 list
在 execute_read/write 回调中执行副作用	移到回调外——回调可能会重试
将 dataclass/Pydantic 对象作为参数传递	先转换为 dict
UNWIND 使用对象列表	仅使用 list[dict]
用 record.get() 检测键是否不存在	键不存在时使用 "key" in record.keys()；.get() 对键不存在和图中的 null 都会返回 None
在 session.run() 后未调用 .consume()	提交时机不确定；请调用 .consume()
在 asyncio 中使用同步驱动程序	使用 AsyncGraphDatabase——同步操作会阻塞事件循环
每个请求都创建异步驱动程序	使用单例——在启动时创建一次
会话未关闭	始终使用 with driver.session(...) as session
对包含时间值的记录调用 json.dumps(record.data())	会触发 TypeError——在 Cypher 中调用 toString() 或进行转换。整个节点/关系可以序列化，但会丢失标签、ID 和关系属性——请投影标量值
对 EagerResult 使用 result["name"]	使用 result.records[0]["name"] 索引，或解包 records, _, _ = ...
认为 Result.single() 在行数为 0 或 2 行以上时会抛出异常	默认 strict=False：返回 None（0 行），或返回第一条记录并发出警告（2 行以上）。single(strict=True) 会抛出异常
在 lambda 上使用 @unit_of_work	使用具名函数
在 ConstraintError 之前捕获 Neo4jError	先捕获 ConstraintError——它是 Neo4jError 的子类
使用 neo4j-driver 作为包名	自 v6 起包名为 neo4j；neo4j-driver 已弃用

---

参考资料

按需加载：

• references/async.md — 完整的异步模式：托管事务、结果方法、并发
• references/data-types.md — 完整的 Python↔Cypher 类型映射、时间类型转换、图对象 API、空间类型（CartesianPoint/WGS84Point）
• references/performance.md — 连接池、惰性流式处理、线程与 asyncio、书签/因果一致性
• references/transactions.md — 显式事务、回滚、提交不确定性、unit_of_work 详情

文档：

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

---

检查清单

• 已将包安装为 neo4j（而非 neo4j-driver）
• 在启动时创建一个 Driver 实例，并在各处共享
• 在启动时调用 verify_connectivity()
• 在每次调用中设置 database_ / database=
• 使用 $param 占位符——不使用 f-string 或 .format()
• 在事务回调中读取结果（不直接返回原始结果）
• 将会话用作上下文管理器（with driver.session(...) as session）
• 在 Neo4jError 之前捕获 ConstraintError
• 在 asyncio 代码中使用 AsyncGraphDatabase（而非同步驱动程序）
• 在应用启动时创建一次异步驱动程序（而非每个请求都创建）
• 在 execute_read/write 回调之外执行副作用
• UNWIND 批次使用 list[dict]