本文目录导读：
我来详细介绍Python操作图数据库的方法,以最流行的Neo4j为例，同时也涵盖其他图数据库。
使用Neo4j（最常用）
安装驱动
pip install neo4j
基本连接和操作
from neo4j import GraphDatabase
class Neo4jConnection:
def __init__(self, uri, user, password):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def close(self):
self.driver.close()
def create_person(self, name, age):
with self.driver.session() as session:
result = session.execute_write(
self._create_person_tx, name, age
)
return result
@staticmethod
def _create_person_tx(tx, name, age):
query = (
"CREATE (p:Person {name: $name, age: $age}) "
"RETURN p"
)
result = tx.run(query, name=name, age=age)
return result.single()[0]
def create_relationship(self, person1, person2, relation):
with self.driver.session() as session:
session.execute_write(
self._create_relationship_tx, person1, person2, relation
)
@staticmethod
def _create_relationship_tx(tx, person1, person2, relation):
query = (
"MATCH (a:Person {name: $name1}) "
"MATCH (b:Person {name: $name2}) "
"CREATE (a)-[r:%s]->(b)" % relation
)
tx.run(query, name1=person1, name2=person2)
# 使用示例
conn = Neo4jConnection("bolt://localhost:7687", "neo4j", "password")
# 创建节点
conn.create_person("Alice", 30)
conn.create_person("Bob", 25)
# 创建关系
conn.create_relationship("Alice", "Bob", "KNOWS")
conn.close()
查询数据
def query_relationships(self, person_name):
with self.driver.session() as session:
result = session.run(
"MATCH (p:Person {name: $name})-[r]->(friend) "
"RETURN p.name, type(r), friend.name",
name=person_name
)
for record in result:
print(f"{record['p.name']} - {record['type(r)']} - {record['friend.name']}")
使用py2neo（Neo4j的ORM）
安装
pip install py2neo
基本操作
from py2neo import Graph, Node, Relationship
# 连接图数据库
graph = Graph("bolt://localhost:7687", auth=("neo4j", "password"))
# 创建节点
alice = Node("Person", name="Alice", age=30)
bob = Node("Person", name="Bob", age=25)
graph.create(alice)
graph.create(bob)
# 创建关系
knows = Relationship(alice, "KNOWS", bob)
graph.create(knows)
# 查询
result = graph.run("MATCH (p:Person) RETURN p.name, p.age")
for record in result:
print(f"Name: {record['p.name']}, Age: {record['p.age']}")
# 使用py2neo的查询语法
from py2neo import NodeMatcher
matcher = NodeMatcher(graph)
persons = matcher.match("Person", name="Alice").first()
print(persons)
操作其他图数据库
ArangoDB
pip install python-arango
from arango import ArangoClient
# 连接
client = ArangoClient(hosts='http://localhost:8529')
db = client.db('mydb', username='root', password='password')
# 创建集合
if not db.has_graph('social'):
graph = db.create_graph('social')
else:
graph = db.graph('social')
# 创建顶点集合
if not graph.has_vertex_collection('persons'):
persons = graph.create_vertex_collection('persons')
# 创建边集合
if not graph.has_edge_definition('knows'):
graph.create_edge_definition(
edge_collection='knows',
from_vertex_collections=['persons'],
to_vertex_collections=['persons']
)
# 插入数据
persons.insert({'_key': 'alice', 'name': 'Alice', 'age': 30})
persons.insert({'_key': 'bob', 'name': 'Bob', 'age': 25})
# 创建边
graph.create_edge('knows', {'_from': 'persons/alice', '_to': 'persons/bob'})
JanusGraph（通过Gremlin）
pip install gremlinpython
from gremlin_python import statics
from gremlin_python.structure.graph import Graph
from gremlin_python.process.graph_traversal import __
from gremlin_python.process.strategies import *
from gremlin_python.driver.driver_remote_connection import DriverRemoteConnection
# 连接
graph = Graph()
g = graph.traversal().withRemote(
DriverRemoteConnection('ws://localhost:8182/gremlin', 'g')
)
# 添加顶点
alice = g.addV('person').property('name', 'Alice').property('age', 30).next()
bob = g.addV('person').property('name', 'Bob').property('age', 25).next()
# 添加边
g.addE('knows').from_(alice).to(bob).next()
# 查询
result = g.V().has('name', 'Alice').out('knows').values('name').toList()
print(result) # ['Bob']
高级操作示例
批量插入
def batch_create_nodes(self, nodes_data):
"""批量创建节点"""
with self.driver.session() as session:
# 使用UNWIND进行批量操作
session.run(
"""
UNWIND $nodes AS node
CREATE (p:Person {
name: node.name,
age: node.age,
email: node.email
})
""",
nodes=nodes_data
)
# 使用
nodes = [
{"name": "Charlie", "age": 28, "email": "charlie@example.com"},
{"name": "David", "age": 35, "email": "david@example.com"},
]
batch_create_nodes(nodes)
复杂查询
def find_shortest_path(self, start_name, end_name):
"""查找最短路径"""
with self.driver.session() as session:
result = session.run(
"""
MATCH (start:Person {name: $start_name}),
(end:Person {name: $end_name}),
p = shortestPath((start)-[*..15]-(end))
RETURN p
""",
start_name=start_name,
end_name=end_name
)
for record in result:
return record['p']
使用索引
def create_indexes(self):
"""创建索引以提高查询性能"""
with self.driver.session() as session:
# 创建单属性索引
session.run("CREATE INDEX person_name FOR (n:Person) ON (n.name)")
# 创建复合索引
session.run("CREATE INDEX person_name_age FOR (n:Person) ON (n.name, n.age)")
最佳实践
连接池管理
from neo4j import GraphDatabase
from contextlib import contextmanager
class Neo4jClient:
def __init__(self, uri, user, password):
self.driver = GraphDatabase.driver(
uri,
auth=(user, password),
max_connection_pool_size=50,
connection_acquisition_timeout=60
)
@contextmanager
def get_session(self):
session = self.driver.session()
try:
yield session
finally:
session.close()
def execute_query(self, query, params=None):
with self.get_session() as session:
result = session.run(query, params)
return [record.data() for record in result]
# 使用
client = Neo4jClient("bolt://localhost:7687", "neo4j", "password")
data = client.execute_query("MATCH (p:Person) RETURN p.name LIMIT 10")
错误处理
from neo4j.exceptions import ServiceUnavailable, AuthError
def safe_operation(self, operation, *args, **kwargs):
try:
return operation(*args, **kwargs)
except ServiceUnavailable:
print("数据库连接失败，请检查服务是否运行")
except AuthError:
print("认证失败，请检查用户名和密码")
except Exception as e:
print(f"操作失败: {e}")
- Neo4j：最流行，文档丰富，推荐优先使用
- py2neo：适合不想写Cypher语句的场景
- ArangoDB：支持多模型，灵活性高
- JanusGraph：适合大规模图数据，使用Gremlin查询
选择哪种方案取决于你的具体需求、数据规模和团队熟悉度。
