neo4j-driver-python-skill

使用する場面

• Neo4j に接続する Python コードを書く場合
• ドライバー、セッション、トランザクション、または非同期パターンをセットアップする場合
• 結果の処理、シリアライズ、または UNWIND によるバッチ処理の問題をデバッグする場合
• Python コードでの Neo4j ドライバーの使用方法をレビューする場合

使用しない場面

• Cypher の作成／最適化 → neo4j-cypher-skill
• ドライバーのバージョンアップグレード → neo4j-migration-skill
• GraphRAG パイプライン（neo4j-graphrag パッケージ） → 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 コンテキストは [6.3+] で SSLKEYLOGFILE を尊重します。Python 3.15 は [6.4+] でサポートされています。neo4j>=6.4.1 に固定してください。6.4.0 ではプール取得が無限に続くことがあり、HTTP Query API でベアラートークンが二重にエンコードされます。

Neo4j 2026.08+ の UUID プロパティを uuid.UUID としてラウンドトリップするには、neo4j>=6.3 が必要です。UUID の保存にはブロックストア形式（Enterprise）が必要です。Community（アライン形式）では失敗し、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 を使用しない場合は、シェルで export を使うか、os.getenv を直接使用してください。

---

ドライバーのライフサイクル

アプリケーションごとに Driver を1つ作成してください。スレッドセーフですが、作成コストが高いため、リクエストごとに作成しないでください。

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/）	Bolt の代わりに HTTP Query API を使用 [6.4+、プレビュー] — pip install "neo4j[http]"

認証オプション：("user", "pass") タプル、basic_auth()、bearer_auth("jwt")、kerberos_auth("b64")。

HTTP Query API [6.4+、プレビュー — セマンティックバージョニングの保証なし]：すべての呼び出しで database_ / database を渡してください（ホームデータベースの解決なし）。認証は basic（realm なし）または bearer のみです。トランザクションメタデータとタイムアウト、fetch_size（バックプレッシャーなし）、通知フィルターは無視されるか、サポートされません。VECTOR、UUID、UnsupportedType は交換できません。これらを使用する場合は Bolt が必要です。neo4j+s:// / bolt+s:// を使用してください。

---

適切な API の選択

API	使用する場面	自動再試行	ストリーミング
driver.execute_query()	ほとんどのクエリ — シンプルで安全なデフォルト	✅	❌ 一括取得
session.execute_read/write()	大量の結果／1つのトランザクションで複数のクエリを実行する場合	✅	✅
session.run()	LOAD CSV、CALL {} IN TRANSACTIONS、スクリプト	⚠️ 1回のみ [6.2+]	✅
AsyncGraphDatabase	asyncio アプリケーション	✅	✅

session.run() の再試行 [6.2+]：DBMS が冪等とマークしたエラー（現在はアドミッション制御）の場合に限り、直ちに1回再試行します。ドライバーまたはセッションレベルで 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件以上 の場合は最初のレコードと警告が返されます（例外は発生しません）。ResultNotSingleError が発生するのは single(strict=True) の場合のみです（0件または2件以上）。必要な行がちょうど1件の場合（例：result_transformer_=lambda r: r.single(strict=True)）は strict=True を使用してください。それ以外の場合は None かどうかを確認してください。

管理トランザクション（execute_read / execute_write）

大きな結果や、1つのトランザクション内で複数のクエリを実行する場合に使用します。

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 によるタイムアウト／メタデータの設定（名前付き関数のみ。ラムダにはデコレータを適用できません）：

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 がマークした）エラーに対してのみ、1回だけ即時に再試行します［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 のサブクラスであり、先に捕捉しないと後者に捕捉されてしまうため、Neo4jError より先に捕捉してください。

---

結果へのアクセスと null 安全性

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 → プロパティの dict、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 = 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 を使用してください。
• 一括書き込みでは、リスト全体を1つの 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,
)

セッションの枯渇：開いているセッションはそれぞれ1つの接続を保持します。必ず 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 です。0 件の場合は None、2 件以上の場合は最初のレコードと警告を返します。single(strict=True) は例外を発生させます
ラムダに @unit_of_work を付ける	名前付き関数を使用します
ConstraintError より先に Neo4jError を捕捉する	ConstraintError はサブクラスなので、先に捕捉します
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 インスタンスを 1 つ作成し、全体で共有する
• 起動時に verify_connectivity() を呼び出す
• すべての呼び出しで database_ / database= を設定する
• $param プレースホルダーを使用する（f-string や .format() は使わない）
• トランザクションコールバック内で結果を取得する（未加工のまま返さない）
• セッションをコンテキストマネージャーとして使用する（with driver.session(...) as session）
• Neo4jError より先に ConstraintError を捕捉する
• asyncio コードでは AsyncGraphDatabase を使用する（同期ドライバーは使わない）
• 非同期ドライバーはアプリの起動時に一度だけ作成する（リクエストごとには作成しない）
• execute_read/write コールバックの外で副作用を実行する
• UNWIND のバッチには list[dict] を使用する