如何让Neptune的make_signed_request返回字典列表格式的GraphSON数据？

问题背景 #

使用AWS官方提供的签名请求代码执行Gremlin查询时：

make_signed_request(query="g.V().limit(10).valueMap(true).toList()")

返回的是包含@type、@value的原始GraphSON结构，无法直接使用；但在Neptune Notebook中通过%%gremlin --store-to foo执行同一查询后，变量foo是已经格式化好的字典列表。需要让Python脚本中的签名请求返回一致的格式化结果。

解决方案 #

Neptune Notebook的%%gremlin魔法命令底层通过Gremlin Python SDK的GraphSON解析器自动处理了结果转换，你可以在脚本中手动实现这一逻辑：

方法1：手动解析GraphSON结果 #

1. 先安装Gremlin Python依赖：

pip install gremlinpython

2. 导入解析器并处理返回数据：

from gremlin_python.structure.io.graphsonV3d0 import GraphSONReader

# 获取原始GraphSON响应（假设make_signed_request返回的是JSON字符串或已加载的字典）
raw_graphson = make_signed_request(query="g.V().limit(10).valueMap(true).toList()")

# 使用GraphSONReader解析成Python原生数据结构
reader = GraphSONReader()
formatted_result = reader.read(raw_graphson)

解析后的formatted_result就是和Notebook中foo一致的字典列表。

方法2：直接使用Gremlin Python客户端（更推荐） #

如果不需要手动维护签名逻辑，直接使用Gremlin Python客户端连接Neptune并启用IAM认证，它会自动处理请求签名和结果解析：

from gremlin_python.driver.driver_remote_connection import DriverRemoteConnection
from gremlin_python.process.anonymous_traversal import traversal
from gremlin_python.driver.aiohttp.transport import AiohttpTransport
from gremlin_python.driver.serializer import GraphSONSerializersV3d0
import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
import datetime

# 生成IAM签名（替换为你的AWS区域和Neptune端点）
region = "us-east-1"
neptune_endpoint = "your-neptune-cluster-endpoint:8182"
service = "neptune-db"

session = boto3.Session()
credentials = session.get_credentials()
current_credentials = credentials.get_frozen_credentials()

request = AWSRequest(
    method="GET",
    url=f"https://{neptune_endpoint}/gremlin",
    headers={"Content-Type": "application/json"}
)

sigv4 = SigV4Auth(current_credentials, service, region)
sigv4.add_auth(request)
auth_header = request.headers["Authorization"]

# 建立连接并执行查询
conn = DriverRemoteConnection(
    f"wss://{neptune_endpoint}/gremlin",
    "g",
    transport_factory=lambda: AiohttpTransport(
        headers={
            'Authorization': auth_header,
            'X-Amz-Date': datetime.datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')
        },
        serializer=GraphSONSerializersV3d0()
    )
)

g = traversal().withRemote(conn)
formatted_result = g.V().limit(10).valueMap(true).toList()

# 关闭连接
conn.close()

这种方式不仅省去了手动解析的步骤，还能让你直接使用Gremlin Python的遍历API，代码更简洁易维护。

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