Python脚本如何操作图数据库

本文目录导读：

1. 使用Neo4j（最常用）
2. 使用py2neo（Neo4j的ORM）
3. 操作其他图数据库
4. 高级操作示例
5. 最佳实践

我来详细介绍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查询

选择哪种方案取决于你的具体需求、数据规模和团队熟悉度。