AWS Lambda调用Amazon Neptune文档示例代码报错求助

问题背景 #

依据Amazon Neptune官方文档示例创建AWS Lambda函数，已配置好所有必需的环境变量，但测试时触发TypeError，错误提示为'GraphTraversal' object is not callable。

完整代码 #

import os, sys, backoff, math
from random import randint
from gremlin_python import statics
from gremlin_python.driver.driver_remote_connection import DriverRemoteConnection
from gremlin_python.driver.protocol import GremlinServerError
from gremlin_python.driver import serializer
from gremlin_python.process.anonymous_traversal import traversal
from gremlin_python.process.graph_traversal import __
from gremlin_python.process.strategies import *
from gremlin_python.process.traversal import T
from tornado.websocket import WebSocketClosedError
from tornado import httpclient 
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.credentials import ReadOnlyCredentials
from types import SimpleNamespace

reconnectable_err_msgs = [ 
  'ReadOnlyViolationException',
  'Server disconnected',
  'Connection refused'
]

retriable_err_msgs = ['ConcurrentModificationException'] + reconnectable_err_msgs

network_errors = [WebSocketClosedError, OSError]

retriable_errors = [GremlinServerError] + network_errors    

def prepare_iamdb_request(database_url):
    
  service = 'neptune-db'
  method = 'GET'

  access_key = os.environ['AWS_ACCESS_KEY_ID']
  secret_key = os.environ['AWS_SECRET_ACCESS_KEY']
  region = os.environ['AWS_REGION']
  session_token = os.environ['AWS_SESSION_TOKEN']
  
  creds = SimpleNamespace(
    access_key=access_key, secret_key=secret_key, token=session_token, region=region,
  )

  request = AWSRequest(method=method, url=database_url, data=None)
  SigV4Auth(creds, service, region).add_auth(request)
  
  return httpclient.HTTPRequest(database_url, headers=request.headers.items())
    
def is_retriable_error(e):

  is_retriable = False
  err_msg = str(e)
  
  if isinstance(e, tuple(network_errors)):
    is_retriable = True
  else:
    is_retriable = any(retriable_err_msg in err_msg for retriable_err_msg in retriable_err_msgs)
  
  print('error: [{}] {}'.format(type(e), err_msg))
  print('is_retriable: {}'.format(is_retriable))
  
  return is_retriable

def is_non_retriable_error(e):    
  return not is_retriable_error(e)
    
def reset_connection_if_connection_issue(params):
  
  is_reconnectable = False

  e = sys.exc_info()[1]
  err_msg = str(e)
  
  if isinstance(e, tuple(network_errors)):
    is_reconnectable = True
  else:
    is_reconnectable = any(reconnectable_err_msg in err_msg for reconnectable_err_msg in reconnectable_err_msgs)
    
  print('is_reconnectable: {}'.format(is_reconnectable))
    
  if is_reconnectable:
    global conn
    global g
    conn.close()
    conn = create_remote_connection()
    g = create_graph_traversal_source(conn)
   
@backoff.on_exception(backoff.constant,
  tuple(retriable_errors),
  max_tries=5,
  jitter=None,
  giveup=is_non_retriable_error,
  on_backoff=reset_connection_if_connection_issue,
  interval=1)
def query(**kwargs):
  
  id = kwargs['id']
  
  return (g.V(id)
    .fold()
    .coalesce(
      __.unfold(), 
      __.addV('User').property(T.id, id)
    )
    .id().next())
    
def doQuery(event):
  return query(id=str(randint(0, 10000)))

def lambda_handler(event, context):
  return doQuery(event)
  
def create_graph_traversal_source(conn):
  return traversal().withRemote(conn)
  
def create_remote_connection():
  print('Creating remote connection')
  
  return DriverRemoteConnection(
    connection_string(),
    'g',
    pool_size=1,
    message_serializer=serializer.GraphSONSerializersV2d0())
  
def connection_string():
  
  database_url = 'wss://{}:{}/gremlin'.format(os.environ['neptuneEndpoint'], os.environ['neptunePort'])
  
  if 'USE_IAM' in os.environ and os.environ['USE_IAM'] == 'true':
    return prepare_iamdb_request(database_url)
  else:
    return database_url
  
conn = create_remote_connection()
g = create_graph_traversal_source(conn)

错误信息 #

{
  "errorMessage": "'GraphTraversal' object is not callable",
  "errorType": "TypeError",
  "requestId": "69e6ecd3-1291-4d21-a8fa-1fc910525fc1",
  "stackTrace": [
    "  File \"/var/task/lambda_function.py\", line 111, in lambda_handler\n    return doQuery(event)\n",
    "  File \"/var/task/lambda_function.py\", line 108, in doQuery\n    return query(id=str(randint(0, 10000)))\n",
    "  File \"/var/task/backoff/_sync.py\", line 105, in retry\n    ret = target(*args, **kwargs)\n",
    "  File \"/var/task/lambda_function.py\", line 99, in query\n    return (g.V(id)\n"
  ]
}

日志输出 #

LOGS    Name: cloudwatch_lambda_agent   State: Subscribed   Types: [Platform]
Creating remote connection
EXTENSION   Name: cloudwatch_lambda_agent   State: Ready    Events: [INVOKE,SHUTDOWN]
START RequestId: 69e6ecd3-1291-4d21-a8fa-1fc910525fc1 Version: $LATEST
[ERROR] TypeError: 'GraphTraversal' object is not callable
Traceback (most recent call last):
  File "/var/task/lambda_function.py", line 111, in lambda_handler
    return doQuery(event)
  File "/var/task/lambda_function.py", line 108, in doQuery
    return query(id=str(randint(0, 10000)))
  File "/var/task/backoff/_sync.py", line 105, in retry
    ret = target(*args, **kwargs)
  File "/var/task/lambda_function.py", line 99, in query
    return (g.V(id)END RequestId: 69e6ecd3-1291-4d21-a8fa-1fc910525fc1
REPORT RequestId: 69e6ecd3-1291-4d21-a8fa-1fc910525fc1  Duration: 108.64 ms Billed Duration: 109 ms Memory Size: 128 MB Max Memory Used: 88 MB  Init Duration: 1025.72 ms   

错误原因 #

核心问题是全局变量g的类型异常：

• Lambda冷启动时，全局初始化的g本应是GraphTraversalSource（可调用的遍历源），但如果连接创建过程中出现隐性异常（如IAM认证失败、网络超时未抛出致命错误），会导致g变成GraphTraversal对象（不可调用）。
• Lambda执行环境复用（热启动）时，全局变量会被保留，若之前的连接已断开，g会成为无效的遍历对象，而非重新初始化。此时调用g.V(id)就会触发"对象不可调用"的错误。

解决方案 #

修改全局变量初始化逻辑 #

1. 移除末尾的全局初始化代码：

# 删除这两行
# conn = create_remote_connection()
# g = create_graph_traversal_source(conn)

2. 在lambda_handler中初始化全局变量，并增加有效性检查：

def doQuery(event):
    global conn, g
    # 验证连接有效性，无效则重建
    try:
        g.V().limit(1).next()
    except Exception as e:
        print(f"Connection invalid, reconnecting: {e}")
        conn = create_remote_connection()
        g = create_graph_traversal_source(conn)
    return query(id=str(randint(0, 10000)))

def lambda_handler(event, context):
    global conn, g
    # 首次初始化全局连接和遍历源
    if 'conn' not in globals() or conn is None:
        conn = create_remote_connection()
        g = create_graph_traversal_source(conn)
    return doQuery(event)

优化重连逻辑 #

确保reset_connection_if_connection_issue函数能正确更新全局变量：

def reset_connection_if_connection_issue(params):
    is_reconnectable = False
    e = sys.exc_info()[1]
    err_msg = str(e)
    
    if isinstance(e, tuple(network_errors)):
        is_reconnectable = True
    else:
        is_reconnectable = any(reconnectable_err_msg in err_msg for reconnectable_err_msg in reconnectable_err_msgs)
    
    print(f'is_reconnectable: {is_reconnectable}')
    
    if is_reconnectable:
        global conn, g
        try:
            conn.close()
        except:
            pass  # 忽略关闭失败的情况
        conn = create_remote_connection()
        g = create_graph_traversal_source(conn)

官方文档示例失效的原因 #

官方示例未考虑Lambda的执行环境复用机制：

• 示例假设全局初始化的连接和遍历源始终有效，但实际生产环境中，网络波动、IAM临时凭证过期、Neptune节点重启等情况都会导致连接失效。
• 当Lambda执行环境被复用时，全局变量不会重新初始化，失效的g会变成不可调用的GraphTraversal对象，触发错误。
• 示例缺少连接有效性检查和自动重连的完整逻辑，无法适配生产环境的复杂场景。

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