如何通过部署在AWS Lambda上的FastAPI调用已有Lambda函数?
可以在FastAPI(通过Mangum部署到Lambda)中创建路由调用已有Lambda函数吗?
当然可以实现。你只需要借助AWS官方的Python SDK boto3 发起对目标Lambda函数的调用,同时要确保当前运行FastAPI的Lambda执行角色拥有调用目标Lambda的权限。
具体实现步骤:
准备依赖:
Lambda运行环境默认自带boto3,如果你的项目需要特定版本,或者要打包完整依赖,只需在requirements.txt中添加相关包:fastapi>=0.100.0 mangum>=0.17.0 boto3>=1.28.0配置IAM权限:
给当前FastAPI所在Lambda的执行角色添加权限,允许它调用目标Lambda函数。示例权限策略如下:{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "lambda:InvokeFunction", "Resource": "arn:aws:lambda:你的AWS区域:你的账号ID:function:目标Lambda函数名称" } ] }编写FastAPI路由代码:
在你的FastAPI应用中新增路由,通过boto3调用目标Lambda。示例代码如下:from fastapi import FastAPI from mangum import Mangum import boto3 import json from botocore.exceptions import ClientError app = FastAPI() # 初始化Lambda客户端 lambda_client = boto3.client('lambda') @app.get("/") def root(): return {"message": "Up and running!"} @app.post("/call-target-lambda") def call_target_lambda(function_name: str, payload: dict): try: # 同步调用目标Lambda(等待执行结果返回) response = lambda_client.invoke( FunctionName=function_name, Payload=json.dumps(payload).encode('utf-8'), InvocationType='RequestResponse' ) # 解析目标Lambda返回的结果 result = json.load(response['Payload']) return {"status": "success", "data": result} except ClientError as e: return {"status": "error", "message": str(e)} # 如果需要异步调用(不等待结果),可以修改InvocationType为'Event' @app.post("/async-call-target-lambda") def async_call_target_lambda(function_name: str, payload: dict): try: lambda_client.invoke( FunctionName=function_name, Payload=json.dumps(payload).encode('utf-8'), InvocationType='Event' ) return {"status": "success", "message": "目标Lambda已异步触发"} except ClientError as e: return {"status": "error", "message": str(e)} handler = Mangum(app)
注意事项:
- 同步调用(
RequestResponse)会等待目标Lambda执行完成,适合需要获取执行结果的场景;异步调用(Event)仅触发执行,不等待结果,适合无需即时返回的任务。 - 打包ZIP部署时,若依赖包不在Lambda默认环境中,需将它们一起打包进去;
boto3默认存在,可根据需求选择是否包含。
内容的提问来源于stack exchange,提问作者haju
相关产品推荐
相关产品推荐

