使用 REST API 构建查询服务

通过 REST API 和 FastAPI 创建电影演员查询服务。应用接收电影编号，调用悦数图数据库开发版接口，再返回电影及其参演演员的 JSON 数据。

REST API 为预览功能，仅供评估和测试。接口参考见 REST API。

前提条件

• 已按准备悦数图数据库开发版启动悦数图数据库开发版，并加载movie示例数据集。
• 已安装 Python 3.10 或更高版本，了解函数、字典和异常处理。
• 已准备悦数图数据库开发版用户名和密码，应用所在电脑能够访问悦数图数据库开发版的 HTTP 端口。

以下终端命令适用于 macOS 和 Linux，均在运行应用的电脑上执行。该电脑可以是部署 Docker 的开发机，也可以是能够访问悦数图数据库开发版的另一台电脑。

1. 创建项目并安装依赖

mkdir movie-rest-app
cd movie-rest-app
python3 -m venv .venv
source .venv/bin/activate
python -m pip install httpx==0.28.1 fastapi==0.142.1 uvicorn==0.54.0

HTTPX 向悦数图数据库开发版发送 HTTP 请求，FastAPI 提供应用 HTTP 接口，Uvicorn 运行应用。

2. 配置悦数图数据库开发版连接

按应用所在位置选择悦数图数据库开发版地址：

应用所在位置	悦数图数据库开发版地址
与 Docker 在同一台开发机	http://127.0.0.1:19669
在另一台电脑	http://<开发机IP>:19669

将<开发机IP>替换为运行 Docker 的机器 IP。如果修改过19669:19669端口映射，使用冒号左侧的开发机端口。

在创建项目时使用的终端中设置环境变量。下方以应用与 Docker 在同一台开发机为例：

export NEBULA_HTTP_URL='http://127.0.0.1:19669'
export NEBULA_USER='root'
export NEBULA_PASSWORD="$(python -c 'import getpass; print(getpass.getpass("Database password: "))')"

出现Database password:提示后，输入悦数图数据库开发版当前的实际密码。Compose 文件中用$$表示的字符，在这里输入单个$。后续启动命令继续在此终端执行。

NEBULA_HTTP_URL填写悦数图数据库开发版 HTTP 接口的根地址，不包含/gql或/ui/console。配套配置使用 HTTP。

3. 保存应用代码

下载 movie_rest.py，保存到movie-rest-app目录中。

示例固定查询/default_schema/movie。如果数据位于其他 Schema，修改文件中两条查询的图路径。

查看完整代码

"""Run with: python -m uvicorn movie_rest:app --host 127.0.0.1 --port 8100."""

import logging
import os
from typing import Annotated

import httpx
from fastapi import FastAPI, HTTPException, Path

app = FastAPI(title="Movie cast service (REST API)")
logger = logging.getLogger(__name__)
base_url = os.environ.get("NEBULA_HTTP_URL", "http://127.0.0.1:19669").rstrip("/")
username = os.environ.get("NEBULA_USER", "root")
password = os.environ["NEBULA_PASSWORD"]


class DatabaseError(Exception):
    pass


def post(client, path, body=None):
    response = client.post(path, json=body)
    response.raise_for_status()
    payload = response.json()
    if not isinstance(payload, dict) or payload.get("code") != "00000":
        raise DatabaseError("Database returned an unsuccessful response")
    return payload.get("data", {})


def query_rows(client, statement):
    data = post(client, "/gql", {"gql": statement})
    if not isinstance(data, dict):
        raise DatabaseError("Missing query data")
    rows = data.get("tables")
    if not isinstance(rows, list) or not all(isinstance(row, dict) for row in rows):
        raise DatabaseError("Invalid query result table")
    return rows


@app.get("/movies/{movie_id}/actors")
def get_actors(movie_id: Annotated[int, Path(ge=1, le=9223372036854775807)]):
    movie_query = (
        "USE /default_schema/movie MATCH (m@Movie) "
        f"WHERE m.id = {movie_id} RETURN m.name AS movie"
    )
    actors_query = (
        "USE /default_schema/movie MATCH (a@Actor)-[:Act]->(m@Movie) "
        f"WHERE m.id = {movie_id} "
        "RETURN a.id AS actor_id, a.name AS actor ORDER BY actor_id"
    )

    try:
        with httpx.Client(base_url=base_url, timeout=10.0) as client:
            auth = post(client, "/auth", {"username": username, "password": password})
            token = auth.get("token") if isinstance(auth, dict) else None
            if not isinstance(token, str) or not token:
                raise DatabaseError("Missing authentication token")
            client.headers["Authorization"] = f"Bearer {token}"
            try:
                movies = query_rows(client, movie_query)
                actors = query_rows(client, actors_query) if movies else []
            finally:
                # Release this request's database session even if a query fails.
                try:
                    post(client, "/signout")
                except Exception as exc:
                    logger.warning("Signout failed (%s)", type(exc).__name__)
    except Exception as exc:
        logger.warning("Database request failed (%s)", type(exc).__name__)
        raise HTTPException(status_code=502, detail="Database request failed") from exc

    if not movies:
        raise HTTPException(status_code=404, detail="Movie not found")

    return {"movie_id": movie_id, "movie": movies[0]["movie"], "actors": actors}

4. 启动应用

在项目目录中、虚拟环境已激活的终端中执行：

python -m uvicorn movie_rest:app --host 127.0.0.1 --port 8100

如果8100端口已被占用，先停止占用它的应用，或改用--port 8101，并同步修改后续访问地址。

看到以下提示，表示应用已启动：

Application startup complete.
Uvicorn running on http://127.0.0.1:8100 (Press CTRL+C to quit)

保持此终端运行，另开浏览器或终端调用接口。应用监听本机的8100端口；按 Ctrl+C 停止应用。

5. 调用查询接口

在运行应用的同一台电脑上，选择以下任一方式查询编号为3的电影。如果通过 SSH 在远端运行应用，先建立下方的端口转发。

浏览器curl

1. 打开http://127.0.0.1:8100/docs，这是应用自动生成的接口页面。
2. 展开GET /movies/{movie_id}/actors，单击 Try it out。
3. 在movie_id中输入3，单击 Execute，查看响应。

在应用所在电脑另开一个终端，执行：

curl -sS http://127.0.0.1:8100/movies/3/actors | python3 -m json.tool --no-ensure-ascii

两种方式应返回相同的 JSON：

{
  "movie_id": 3,
  "movie": "Shadows in Paradise",
  "actors": [
    {"actor_id": 4826, "actor": "Matti Pellonpää"},
    {"actor_id": 4828, "actor": "Sakari Kuosmanen"},
    {"actor_id": 5999, "actor": "Kati Outinen"}
  ]
}

通过 SSH 访问远端应用

在浏览器所在电脑的本地终端建立转发，将<SSH用户名>和<远端地址>替换为登录应用所在电脑时使用的值：

ssh -N -L 8100:127.0.0.1:8100 <SSH用户名>@<远端地址>

如 SSH 使用非默认端口，添加-p <SSH端口>。认证成功后，转发终端会持续等待，通常没有输出。

保持远端应用终端和本地转发终端运行，在本地浏览器打开http://127.0.0.1:8100/docs。本地8100端口的请求会转发到远端应用的8100端口。

如果本地端口已被占用，将-L后的值改为8101:127.0.0.1:8100，访问地址同步改为http://127.0.0.1:8101/docs。

应用的接口页面位于http://127.0.0.1:8100/docs；悦数图数据库开发版的 API 参考位于http://<开发机IP>:19669/api-docs，其中<开发机IP>为运行 Docker 的机器 IP。

代码说明

查询与响应字段

应用先查询电影名称，再查询参演演员。先检查电影是否存在，可以区分“没有这部电影”和“电影存在但没有演员”：前者返回404，后者返回空的actors数组。

演员查询如下：

USE /default_schema/movie MATCH (a@Actor)-[:Act]->(m@Movie)
WHERE m.id = 3
RETURN a.id AS actor_id, a.name AS actor ORDER BY actor_id

(a@Actor)-[:Act]->(m@Movie)匹配演员参演电影的关系，箭头从演员指向电影。WHERE筛选电影编号，ORDER BY按演员编号排序。应用将示例中的3替换为校验后的整数movie_id。数据模型见示例数据集：movie。

查询结果	应用响应字段
电影查询中的m.name AS movie	movie
演员查询中的a.id AS actor_id	actors中每项的actor_id
演员查询中的a.name AS actor	actors中每项的actor

认证并发送查询

应用为每次请求完成以下调用：

1. 向POST /auth发送悦数图数据库开发版用户名和密码，从data.token读取 JWT。
2. 设置Authorization: Bearer <token>，向POST /gql发送{"gql": "查询语句"}。
3. 检查 HTTP 状态和响应体中的code；成功代码为字符串"00000"，HTTP200的响应也可能包含查询错误。成功时从data.tables读取行数据。
4. 查询结束或失败后，调用POST /signout释放认证会话，再关闭 HTTP 客户端。

post()和query_rows()是完整代码中定义的辅助函数，分别检查响应和读取结果行：

auth = post(client, "/auth", {"username": username, "password": password})
client.headers["Authorization"] = f"Bearer {auth['token']}"
movies = query_rows(client, movie_query)
actors = query_rows(client, actors_query) if movies else []

data.headers列出结果列名，data.tables是行对象数组。查询中的AS固定了字段名，应用可直接读取movie、actor_id和actor。

每条查询都用USE /default_schema/movie指定图，无需先设置会话工作图。多行字符串可直接放入 JSON 的gql字段，无需添加ngql的'''输入标记。

响应状态与排查

HTTP 状态	含义	处理方法
200	查询成功	查看movie和actors；没有演员时数组为空
404	没有这个电影编号	使用数据中已有的编号，例如3
422	电影编号格式或范围不符合要求	使用正整数，且不超过9223372036854775807
502	悦数图数据库开发版 HTTP 请求、认证或查询失败	核对悦数图数据库开发版地址和密码，并按连接并完成首次查询确认数据可访问

需要验证参数校验时，执行：

curl -i http://127.0.0.1:8100/movies/abc/actors

应返回 HTTP422，请求不会发送到悦数图数据库开发版执行查询。应用只接收电影编号，查询结构和图路径由代码固定。

应用无法访问

浏览器显示ERR_CONNECTION_REFUSED或curl无法连接时，检查：

• 应用终端是否仍在运行。
• 请求是否发往应用所在电脑；127.0.0.1始终指运行浏览器或命令的电脑。
• 通过 SSH 访问时，本地转发终端是否仍在运行。

在浏览器所在电脑的终端执行以下命令；如果修改了本地端口，使用修改后的值：

curl -I http://127.0.0.1:8100/docs

应返回HTTP/1.1 200 OK，表示应用的接口页面可以访问。随后再调用电影查询接口，检查悦数图数据库开发版查询结果。

应用终端会记录悦数图数据库开发版请求的异常类型，返回给调用方的信息不包含悦数图数据库开发版密码或 JWT。

扩展练习：查询导演

复制get_actors()函数及其@app.get装饰器，将新函数命名为get_directors()，路由改为/movies/{movie_id}/directors。保留电影存在性检查，将演员查询替换为：

USE /default_schema/movie MATCH (d@Director)-[:Direct]->(m@Movie)
WHERE m.id = 3
RETURN d.id AS director_id, d.name AS director ORDER BY director_id

将演员查询和结果变量改为相应的导演变量。返回 JSON 使用directors数组，其中每项使用director_id和director。

重启应用后调用新接口，核对电影及其导演。电影存在但没有导演关系时，directors为空数组；电影不存在时返回404。可先在悦数图管理控制台中执行上述查询，再与应用结果比较。

继续开发

按 Ctrl+C 停止应用后，悦数图数据库开发版继续运行，图数据保留。更换终端后再次启动时，需重新激活虚拟环境并设置环境变量。

示例每次请求分别认证并登出。自己的应用如需复用 JWT，应处理令牌有效期和认证失败后的更新。悦数图数据库开发版用户权限的配置方法见为用户配置访问权限。

• REST API：查看悦数图数据库开发版请求与响应结构。
• 使用 Python SDK 构建查询服务：通过 SDK 实现相同接口。