在微服务和大数据时代，处理复杂关联数据的需求日益增长。传统的 SQL 数据库在处理多层级关系查询时往往力不从心，而图数据库 Neo4j 凭借其强大的关系遍历能力，成为了知识图谱、社交网络和推荐系统的绝佳选择。今天，我们就来聊聊如何将 FastAPI 的高性能异步特性与 Neo4j 完美结合，打造一个极速响应的图数据接口服务。

🌐 为什么选择 Neo4j 与 FastAPI？

Neo4j 是目前全球最流行的原生图数据库。它摒弃了传统数据库的表结构，采用“节点”和“关系”来存储数据，这种“属性图模型”让它在处理深度关联查询时具备天然优势。无论是金融风控中的反欺诈链路追踪，还是电商系统中的实时商品推荐，Neo4j 都能提供毫秒级的查询体验。

而 FastAPI 作为现代 Python Web 框架的佼佼者，原生支持异步（ASGI）和类型提示。当 FastAPI 遇上 Neo4j 的官方异步驱动（AsyncDriver），我们就能在单线程内通过 asyncio 实现高并发的数据库 I/O 操作，彻底告别传统同步请求的阻塞等待，让接口的吞吐量实现质的飞跃。

🛠️ 环境准备与驱动选择

在开始编码前，我们需要安装必要的依赖库。Neo4j 官方提供了强大的 Python 驱动，完美支持异步操作：

pip install fastapi "uvicorn[standard]" neo4j pydantic

在连接 Neo4j 时，URI 的协议选择至关重要。对于异步驱动，我们通常使用 neo4j://（支持集群路由）或 bolt://（直连）。如果是生产环境且配置了 SSL，则需要使用 neo4j+s:// 或 bolt+s://。

💻 核心代码实现

我们将采用分层架构，将数据库连接封装为独立的 Provider，并在 FastAPI 的路由中调用异步查询。

1. 封装异步 Neo4j 驱动 (neo4j_provider.py)

首先，我们创建一个 Neo4j 的连接管理类。这里利用 AsyncGraphDatabase 创建全局的异步驱动实例，并封装基础的查询方法。

from neo4j import AsyncGraphDatabase, basic_auth

# 替换为你的 Neo4j 实际配置
NEO4J_URI = "neo4j://localhost:7687"
NEO4J_USER = "neo4j"
NEO4J_PASSWORD = "your_password"
NEO4J_DATABASE = "neo4j"

class Neo4jProvider:
    def __init__(self):
        self.driver = AsyncGraphDatabase.driver(
            NEO4J_URI, 
            auth=basic_auth(NEO4J_USER, NEO4J_PASSWORD)
        )

    async def execute_query(self, query: str, parameters: dict = None):
        """执行异步查询并返回序列化后的结果"""
        records, _, _ = await self.driver.execute_query(
            query,
            parameters_=parameters,
            database_=NEO4J_DATABASE,
            routing_="r" # r 代表读请求，w 代表写请求
        )
        # 将 Neo4j 的 Record 对象转换为可序列化的字典
        return [dict(record) for record in records]

    async def close(self):
        await self.driver.close()

# 实例化驱动供全局调用
neo4j_db = Neo4jProvider()

2. 定义数据模型 (schemas.py)

使用 Pydantic 定义接口的请求和响应模型。通过配置 from_attributes = True（Pydantic V2），我们可以轻松地将数据库查询结果映射为标准的 JSON 响应。

from pydantic import BaseModel
from typing import List, Optional

class UserNode(BaseModel):
    id: str
    name: str
    age: Optional[int] = None

    class Config:
        from_attributes = True

class SearchResponse(BaseModel):
    code: int = 200
    message: str = "success"
    data: List[UserNode]

3. 编写 FastAPI 异步路由 (main.py)

最后，我们在 FastAPI 中定义路由。注意这里的路由函数必须使用 async def，并在调用数据库时使用 await。

from fastapi import FastAPI, Query
from fastapi.responses import JSONResponse
from neo4j_provider import neo4j_db
from schemas import SearchResponse, UserNode
import uvicorn

app = FastAPI(title="FastAPI Async Neo4j Demo")

@app.get("/api/users/search", response_model=SearchResponse)
async def search_users(name: str = Query(..., description="用户名称关键字")):
    # 编写 Cypher 查询语句
    cql = """
    MATCH (u:User)
    WHERE u.name CONTAINS $name
    RETURN elementId(u) as id, u.name as name, u.age as age
    """
    
    try:
        # 异步执行查询
        records = await neo4j_db.execute_query(cql, {"name": name})
        
        # 数据映射
        users = [UserNode(**record) for record in records]
        
        return SearchResponse(data=users)
    except Exception as e:
        return JSONResponse(status_code=500, content={"code": 500, "message": str(e)})

# 应用启动与关闭事件
@app.on_event("shutdown")
async def shutdown_event():
    await neo4j_db.close()

if __name__ == "__main__":
    # 注意：host 必须设置为 0.0.0.0 才能支持公网或局域网访问
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

🚀 运行与测试

启动服务后，访问 http://127.0.0.1:8000/docs 即可看到 FastAPI 自动生成的交互式 API 文档。

当我们发起一个带有 name 参数的 GET 请求时，FastAPI 会通过异步驱动向 Neo4j 发送 Cypher 查询。得益于 AsyncDriver 的非阻塞特性，即使在高并发场景下，服务器也能保持极高的响应速度，非常适合用于构建大模型（LLM）的知识图谱检索接口或实时推荐后端。

通过这种全异步的架构，我们不仅充分发挥了 Neo4j 在图计算上的优势，更利用 FastAPI 榨干了 Python 异步编程的性能潜力。希望这篇实战指南能帮助你快速上手，构建出属于自己的高性能图应用！