Flask应用连接Neo4j Aura DB失败问题求助

针对你遇到的「本地单独运行Neo4j查询正常，但Flask服务器调用时出现neo4j.exceptions.ServiceUnavailable: Unable to retrieve routing information」问题，以下是几种常见的解决思路：

1. 确保Neo4j Driver全局复用，避免重复创建 #

单独运行脚本时是单次连接，而Flask处理多请求时，若每次请求都创建新的Driver实例，会导致连接资源耗尽或路由信息获取失败。正确的做法是全局初始化Driver，仅创建一次：

from neo4j import GraphDatabase
from flask import Flask

app = Flask(__name__)
# 全局初始化Driver，仅在应用启动时创建一次
driver = GraphDatabase.driver(
    "neo4j+s://your-instance-id.databases.neo4j.io:7687",
    auth=("neo4j", "your-password")
)

@app.route("/run-query")
def run_query():
    # 每次请求使用Driver创建会话，用完自动关闭
    with driver.session() as session:
        result = session.run("MATCH (n) RETURN count(n) AS node_count")
        return {"node_count": result.single()["node_count"]}

# 应用关闭时关闭Driver，释放资源
@app.teardown_appcontext
def close_driver(exception):
    driver.close()

if __name__ == "__main__":
    app.run()

2. 适配多进程服务器的Driver初始化 #

如果使用Gunicorn等多进程服务器，全局Driver无法跨进程复用，需在每个进程启动后单独初始化Driver：

from neo4j import GraphDatabase
from flask import Flask

app = Flask(__name__)
driver = None

def init_driver():
    """进程内初始化Driver"""
    global driver
    driver = GraphDatabase.driver(
        "neo4j+s://your-instance-id.databases.neo4j.io:7687",
        auth=("neo4j", "your-password")
    )

@app.route("/run-query")
def run_query():
    with driver.session() as session:
        result = session.run("MATCH (n) RETURN count(n) AS node_count")
        return {"node_count": result.single()["node_count"]}

# Gunicorn多进程钩子：每个Worker启动后初始化Driver
def post_fork(server, worker):
    init_driver()

if __name__ == "__main__":
    init_driver()
    app.run()

3. 排查网络/代理差异 #

Flask运行环境与单独脚本的网络配置可能不同（比如IDE代理、系统防火墙规则），导致无法访问Neo4j Aura的7687端口。可以在Flask中添加网络连通性测试：

import socket

@app.route("/test-connection")
def test_connection():
    try:
        # 测试连接Neo4j Aura实例
        sock = socket.create_connection(("your-instance-id.databases.neo4j.io", 7687), timeout=10)
        sock.close()
        return {"status": "connection successful"}
    except Exception as e:
        return {"status": "connection failed", "error": str(e)}

如果测试失败，检查防火墙是否允许Flask进程访问7687端口，或是否需要配置代理。

4. 处理证书验证问题 #

虽然neo4j+s://协议会自动处理证书，但部分环境下可能存在证书链不完整的情况。可以临时禁用证书验证（仅用于测试，生产环境不推荐）：

driver = GraphDatabase.driver(
    "neo4j+s://your-instance-id.databases.neo4j.io:7687",
    auth=("neo4j", "your-password"),
    trust_all_certificates=True
)

生产环境建议下载Neo4j Aura的官方证书，指定证书路径：

driver = GraphDatabase.driver(
    "neo4j+s://your-instance-id.databases.neo4j.io:7687",
    auth=("neo4j", "your-password"),
    trusted_certificates="/path/to/aura-certificate.crt"
)

内容的提问来源于stack exchange，提问作者Salvatore Pennisi