# Python连接Neo4j的完整教程和代码示例 ## 1. 环境准备与驱动选择 在使用Python连接Neo4j之前，需要先安装相应的驱动库。目前主要有两种选择： | 驱动库 | 特点 | 适用场景 | |--------|------|----------| | 官方neo4j驱动 | 官方维护，性能稳定 | 生产环境，需要官方支持 | | py2neo库 | 高级封装，使用简便 | 快速开发，学习阶段 | ### 安装必要的库 ```python # 安装官方neo4j驱动 pip install neo4j # 或者安装py2neo库 pip install py2neo # 如果需要数据可视化，还可以安装相关库 pip install matplotlib networkx ``` ## 2. 使用官方neo4j驱动连接 官方驱动提供了最直接和稳定的连接方式，以下是详细的连接示例： ```python from neo4j import GraphDatabase class Neo4jConnection: def __init__(self, uri, user, password): """ 初始化Neo4j连接 :param uri: Neo4j数据库URI :param user: 用户名 :param password: 密码 """ self._uri = uri self._user = user self._password = password self._driver = None self._connect() def _connect(self): """创建数据库连接""" try: self._driver = GraphDatabase.driver( self._uri, auth=(self._user, self._password) ) # 测试连接 self._driver.verify_connectivity() print("✅ Neo4j连接成功！") except Exception as e: print(f"❌ 连接失败: {e}") def close(self): """关闭连接""" if self._driver is not None: self._driver.close() print("连接已关闭") def execute_query(self, query, parameters=None): """ 执行Cypher查询 :param query: Cypher查询语句 :param parameters: 查询参数 :return: 查询结果 """ with self._driver.session() as session: result = session.run(query, parameters) return list(result) # 使用示例 if __name__ == "__main__": # 连接参数配置 URI = "bolt://localhost:7687" # 默认Bolt端口 USER = "neo4j" PASSWORD = "your_password" # 替换为实际密码 # 创建连接实例 conn = Neo4jConnection(URI, USER, PASSWORD) # 执行测试查询 test_query = "RETURN 'Hello Neo4j!' AS message" result = conn.execute_query(test_query) for record in result: print(record["message"]) # 关闭连接 conn.close() ``` ## 3. 使用py2neo库连接 py2neo提供了更高级的抽象，使用起来更加便捷： ```python from py2neo import Graph, Node, Relationship class Py2NeoConnection: def __init__(self, uri, user, password): """ 初始化py2neo连接 :param uri: 数据库URI :param user: 用户名 :param password: 密码 """ self.uri = uri self.user = user self.password = password self.graph = None self._connect() def _connect(self): """建立连接""" try: self.graph = Graph( self.uri, auth=(self.user, self.password) ) # 测试连接 self.graph.run("RETURN 1 AS test") print("✅ py2neo连接成功！") except Exception as e: print(f"❌ 连接失败: {e}") def create_sample_data(self): """创建示例数据""" # 创建节点 alice = Node("Person", name="Alice", age=30) bob = Node("Person", name="Bob", age=25) # 创建关系 knows = Relationship(alice, "KNOWS", bob, since=2020) # 提交到数据库 tx = self.graph.begin() tx.create(alice) tx.create(bob) tx.create(knows) tx.commit() print("示例数据创建成功！") def query_data(self): """查询数据""" query = """ MATCH (p:Person) RETURN p.name AS name, p.age AS age """ result = self.graph.run(query) print("查询结果：") for record in result: print(f"姓名: {record['name']}, 年龄: {record['age']}") # 使用示例 if __name__ == "__main__": # 连接配置 URI = "bolt://localhost:7687" USER = "neo4j" PASSWORD = "your_password" # 创建连接 py2neo_conn = Py2NeoConnection(URI, USER, PASSWORD) # 创建示例数据 py2neo_conn.create_sample_data() # 查询数据 py2neo_conn.query_data() ``` ## 4. 连接配置详解 ### 连接参数说明 | 参数 | 说明 | 示例值 | |------|------|--------| | URI格式 | 连接协议和地址 | bolt://host:port | | 默认端口 | Bolt协议默认端口 | 7687 | | HTTP端口 | HTTP API端口 | 7474 | | 认证方式 | 用户名密码认证 | auth=(user, password) | ### 不同环境连接配置 ```python # 本地开发环境配置 LOCAL_CONFIG = { "uri": "bolt://localhost:7687", "user": "neo4j", "password": "neo4j" # 首次安装的默认密码 } # 生产环境配置（可能需要SSL） PRODUCTION_CONFIG = { "uri": "bolt://your-server.com:7687", "user": "admin", "password": "secure_password", "encrypted": True # 启用SSL加密 } # Docker环境配置 DOCKER_CONFIG = { "uri": "bolt://host.docker.internal:7687", "user": "neo4j", "password": "password" } ``` ## 5. 完整实战示例 下面是一个完整的应用示例，展示从连接到基本CRUD操作的全流程： ```python from neo4j import GraphDatabase import json class Neo4jManager: def __init__(self, config): self.driver = GraphDatabase.driver( config["uri"], auth=(config["user"], config["password"]) ) def create_person(self, name, age, city): """创建人员节点""" query = """ CREATE (p:Person {name: $name, age: $age, city: $city}) RETURN p """ parameters = {"name": name, "age": age, "city": city} with self.driver.session() as session: result = session.run(query, parameters) print(f"创建人员: {name}") def create_friendship(self, person1, person2): """创建朋友关系""" query = """ MATCH (a:Person {name: $name1}), (b:Person {name: $name2}) CREATE (a)-[:FRIENDS_WITH {since: date()}]->(b) RETURN a, b """ parameters = {"name1": person1, "name2": person2} with self.driver.session() as session: session.run(query, parameters) print(f"建立朋友关系: {person1} -> {person2}") def find_friends(self, person_name): """查找某人的所有朋友""" query = """ MATCH (p:Person {name: $name})-[:FRIENDS_WITH]->(friend:Person) RETURN friend.name AS friend_name, friend.age AS age, friend.city AS city """ with self.driver.session() as session: result = session.run(query, {"name": person_name}) friends = [] for record in result: friends.append({ "name": record["friend_name"], "age": record["age"], "city": record["city"] }) return friends def close(self): """关闭连接""" self.driver.close() # 主程序 def main(): # 配置连接 config = { "uri": "bolt://localhost:7687", "user": "neo4j", "password": "your_password" } # 创建管理器实例 manager = Neo4jManager(config) try: # 创建示例数据 people = [ ("张三", 28, "北京"), ("李四", 32, "上海"), ("王五", 25, "广州"), ("赵六", 30, "深圳") ] for name, age, city in people: manager.create_person(name, age, city) # 建立朋友关系 friendships = [("张三", "李四"), ("张三", "王五"), ("李四", "赵六")] for person1, person2 in friendships: manager.create_friendship(person1, person2) # 查询张三的朋友 print("\n张三的朋友：") friends = manager.find_friends("张三") for friend in friends: print(f" - {friend['name']} ({friend['age']}岁, {friend['city']})") finally: # 确保连接关闭 manager.close() if __name__ == "__main__": main() ``` ## 6. 常见问题与解决方案 ### 连接失败排查 ```python def diagnose_connection(uri, user, password): """连接诊断函数""" import socket from urllib.parse import urlparse # 解析URI parsed = urlparse(uri) host = parsed.hostname port = parsed.port or 7687 # 检查网络连通性 try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(5) result = sock.connect_ex((host, port)) sock.close() if result == 0: print("✅ 网络连接正常") else: print("❌ 网络连接失败，请检查：") print(" 1. Neo4j服务是否启动") print(" 2. 防火墙设置") print(" 3. 端口是否正确") return False except Exception as e: print(f"❌ 网络检查异常: {e}") return False # 尝试数据库连接 try: from neo4j import GraphDatabase driver = GraphDatabase.driver(uri, auth=(user, password)) driver.verify_connectivity() print("✅ 数据库认证成功") driver.close() return True except Exception as e: print(f"❌ 数据库连接失败: {e}") print("请检查：") print(" 1. 用户名密码是否正确") print(" 2. 数据库是否已初始化") print(" 3. 连接协议是否正确") return False # 使用诊断函数 diagnose_connection("bolt://localhost:7687", "neo4j", "your_password") ``` ### 版本兼容性问题 如果遇到版本兼容性问题，可以尝试以下解决方案： ```python # 检查版本兼容性 import neo4j import py2neo print(f"neo4j驱动版本: {neo4j.__version__}") print(f"py2neo版本: {py2neo.__version__}") # 推荐的版本组合 recommended_versions = { "neo4j": "4.4.0+", "neo4j-driver": "4.4.0+", "py2neo": "2021.2.3+" } ``` 通过以上完整的教程和代码示例，您应该能够成功使用Python连接Neo4j数据库，并进行基本的数据操作。建议从官方驱动开始学习，掌握基础后再根据需求选择是否使用py2neo等高级封装库。