求编写Flask调用AWS API Gateway查询DynamoDB的Lambda请求代码
没问题,我帮你搞定这个Flask应用,用来替代Postman调用你的API Gateway端点。下面是完整的可运行代码,还附带了关键部分的说明,方便你整合到大型应用里:
完整Flask应用代码
from flask import Flask, request, jsonify import requests # 初始化Flask应用 app = Flask(__name__) # 你的API Gateway端点地址 API_GATEWAY_ENDPOINT = "https://3omrrlu81m.execute-api.us-east-1.amazonaws.com/default" def call_api_gateway(payload=None, method="GET"): """封装API Gateway调用逻辑,方便后续复用""" try: if method.upper() == "GET": response = requests.get(API_GATEWAY_ENDPOINT, params=payload) elif method.upper() == "POST": response = requests.post(API_GATEWAY_ENDPOINT, json=payload) # 可以扩展PUT、DELETE等其他HTTP方法 else: return {"error": f"不支持的HTTP方法: {method}"}, 400 # 返回API Gateway的响应内容和状态码 return response.json(), response.status_code except requests.exceptions.RequestException as e: return {"error": f"调用API Gateway失败: {str(e)}"}, 500 # 定义一个路由,模拟Postman发送请求 @app.route('/invoke-dynamodb', methods=['GET', 'POST']) def invoke_dynamodb(): # 获取请求参数(GET用args,POST用json) if request.method == "GET": payload = request.args.to_dict() else: payload = request.get_json() or {} # 调用API Gateway response_data, status_code = call_api_gateway(payload, request.method) # 返回响应给前端 return jsonify(response_data), status_code if __name__ == '__main__': # 运行Flask应用,调试模式可以根据需要关闭 app.run(debug=True)
使用说明
- 安装依赖:首先执行下面的命令安装所需的库:
pip install flask requests - 运行应用:直接执行脚本或者用Flask命令启动:
python app.py # 或者 export FLASK_APP=app.py flask run - 发送请求:
- GET请求:访问
http://localhost:5000/invoke-dynamodb?param1=value1¶m2=value2,参数会自动传递给API Gateway - POST请求:用curl或者其他工具发送JSON payload到
http://localhost:5000/invoke-dynamodb,比如:curl -X POST -H "Content-Type: application/json" -d '{"key": "value"}' http://localhost:5000/invoke-dynamodb
- GET请求:访问
扩展建议
- 如果你的API Gateway需要认证(比如API Key、IAM签名),可以在
call_api_gateway函数里添加对应的请求头或者签名逻辑 - 可以添加日志记录,方便调试和监控请求
- 把API端点配置放到环境变量里,避免硬编码,更适合生产环境
内容的提问来源于stack exchange,提问作者raghusingh
相关产品推荐
相关产品推荐

