AWS SAM Lambda 项目的本地开发服务器-Python教程-PHP中文网

首页

后端开发

Python教程

AWS SAM Lambda 项目的本地开发服务器

Mary-Kate Olsen

Sep 28, 2024 pm 10:10 PM

Local Development Server for AWS SAM Lambda Projects

现在我正在开发一个项目，其中使用 AWS lambda 作为请求处理程序构建 REST API。整个过程使用 AWS SAM 定义 lambda、层并将其连接到漂亮的 template.yaml 文件中的 Api 网关。

问题

本地测试此 API 并不像其他框架那样简单。虽然 AWS 提供了 sam 本地命令来构建托管 lambda 的 Docker 镜像（可以更好地复制 Lambda 环境），但我发现这种方法对于开发过程中的快速迭代来说过于繁重。

解决方案

我想要一种方法：

快速测试我的业务逻辑和数据验证
为前端开发者提供本地服务器进行测试
避免每次更改时重建 Docker 镜像的开销

因此，我创建了一个脚本来满足这些需求。 ?‍♂️

TL;DR：查看此 GitHub 存储库中的 server_local.py。

主要优点

快速设置：启动本地 Flask 服务器，将您的 API 网关路由映射到 Flask 路由。
直接执行：直接触发Python函数（Lambda处理程序），没有Docker开销。
热重载：更改立即反映，缩短开发反馈循环。

此示例基于 sam init 的“Hello World”项目构建，添加了 server_local.py 及其要求以启用本地开发。

阅读 SAM 模板

我在这里所做的是首先阅读 template.yaml，因为我的基础设施和所有 lambda 都有当前定义。

我们创建字典定义所需的所有代码就是这样。为了处理 SAM 模板特有的函数，我向 CloudFormationLoader 添加了一些构造函数。它现在可以支持 Ref 作为对另一个对象的引用、Sub 作为替代方法和 GetAtt 来获取属性。我认为我们可以在这里添加更多逻辑，但现在这完全足以使其工作。

import os
from typing import Any, Dict
import yaml


class CloudFormationLoader(yaml.SafeLoader):
    def __init__(self, stream):
        self._root = os.path.split(stream.name)[0]  # type: ignore
        super(CloudFormationLoader, self).__init__(stream)

    def include(self, node):
        filename = os.path.join(self._root, self.construct_scalar(node))  # type: ignore
        with open(filename, "r") as f:
            return yaml.load(f, CloudFormationLoader)


def construct_getatt(loader, node):
    if isinstance(node, yaml.ScalarNode):
        return {"Fn::GetAtt": loader.construct_scalar(node).split(".")}
    elif isinstance(node, yaml.SequenceNode):
        return {"Fn::GetAtt": loader.construct_sequence(node)}
    else:
        raise yaml.constructor.ConstructorError(
            None, None, f"Unexpected node type for !GetAtt: {type(node)}", node.start_mark
        )


CloudFormationLoader.add_constructor(
    "!Ref", lambda loader, node: {"Ref": loader.construct_scalar(node)}  # type: ignore
)
CloudFormationLoader.add_constructor(
    "!Sub", lambda loader, node: {"Fn::Sub": loader.construct_scalar(node)}  # type: ignore
)
CloudFormationLoader.add_constructor("!GetAtt", construct_getatt)


def load_template() -> Dict[str, Any]:
    with open("template.yaml", "r") as file:
        return yaml.load(file, Loader=CloudFormationLoader)

登录后复制

这会生成这样的 json：

{
   "AWSTemplateFormatVersion":"2010-09-09",
   "Transform":"AWS::Serverless-2016-10-31",
   "Description":"sam-app\nSample SAM Template for sam-app\n",
   "Globals":{
      "Function":{
         "Timeout":3,
         "MemorySize":128,
         "LoggingConfig":{
            "LogFormat":"JSON"
         }
      }
   },
   "Resources":{
      "HelloWorldFunction":{
         "Type":"AWS::Serverless::Function",
         "Properties":{
            "CodeUri":"hello_world/",
            "Handler":"app.lambda_handler",
            "Runtime":"python3.9",
            "Architectures":[
               "x86_64"
            ],
            "Events":{
               "HelloWorld":{
                  "Type":"Api",
                  "Properties":{
                     "Path":"/hello",
                     "Method":"get"
                  }
               }
            }
         }
      }
   },
   "Outputs":{
      "HelloWorldApi":{
         "Description":"API Gateway endpoint URL for Prod stage for Hello World function",
         "Value":{
            "Fn::Sub":"https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/hello/"
         }
      },
      "HelloWorldFunction":{
         "Description":"Hello World Lambda Function ARN",
         "Value":{
            "Fn::GetAtt":[
               "HelloWorldFunction",
               "Arn"
            ]
         }
      },
      "HelloWorldFunctionIamRole":{
         "Description":"Implicit IAM Role created for Hello World function",
         "Value":{
            "Fn::GetAtt":[
               "HelloWorldFunctionRole",
               "Arn"
            ]
         }
      }
   }
}

登录后复制

处理层

这样就可以轻松地为每个端点动态创建 Flask 路由。但在此之前还有一些额外的事情。

在 sam init helloworld 应用程序中没有定义任何层。但我在实际项目中遇到了这个问题。为了使其正常工作，我添加了一个函数来读取层定义并将它们添加到 sys.path 中，以便 python 导入可以正常工作。检查这个：

def add_layers_to_path(template: Dict[str, Any]):
    """Add layers to path. Reads the template and adds the layers to the path for easier imports."""
    resources = template.get("Resources", {})
    for _, resource in resources.items():
        if resource.get("Type") == "AWS::Serverless::LayerVersion":
            layer_path = resource.get("Properties", {}).get("ContentUri")
            if layer_path:
                full_path = os.path.join(os.getcwd(), layer_path)
                if full_path not in sys.path:
                    sys.path.append(full_path)

登录后复制

创建 Flask 路由

在其中我们需要循环遍历资源并找到所有函数。基于此，我正在创建烧瓶路线的数据需求。

def export_endpoints(template: Dict[str, Any]) -> List[Dict[str, str]]:
    endpoints = []
    resources = template.get("Resources", {})
    for resource_name, resource in resources.items():
        if resource.get("Type") == "AWS::Serverless::Function":
            properties = resource.get("Properties", {})
            events = properties.get("Events", {})
            for event_name, event in events.items():
                if event.get("Type") == "Api":
                    api_props = event.get("Properties", {})
                    path = api_props.get("Path")
                    method = api_props.get("Method")
                    handler = properties.get("Handler")
                    code_uri = properties.get("CodeUri")

                    if path and method and handler and code_uri:
                        endpoints.append(
                            {
                                "path": path,
                                "method": method,
                                "handler": handler,
                                "code_uri": code_uri,
                                "resource_name": resource_name,
                            }
                        )
    return endpoints

登录后复制

下一步就是使用它并为每个人设置一条路线。

def setup_routes(template: Dict[str, Any]):
    endpoints = export_endpoints(template)
    for endpoint in endpoints:
        setup_route(
            endpoint["path"],
            endpoint["method"],
            endpoint["handler"],
            endpoint["code_uri"],
            endpoint["resource_name"],
        )


def setup_route(path: str, method: str, handler: str, code_uri: str, resource_name: str):
    module_name, function_name = handler.rsplit(".", 1)
    module_path = os.path.join(code_uri, f"{module_name}.py")
    spec = importlib.util.spec_from_file_location(module_name, module_path)
    if spec is None or spec.loader is None:
        raise Exception(f"Module {module_name} not found in {code_uri}")
    module = importlib.util.module_from_spec(spec)

    spec.loader.exec_module(module)
    handler_function = getattr(module, function_name)

    path = path.replace("{", "<").replace("}", ">")

    print(f"Setting up route for [{method}] {path} with handler {resource_name}.")

    # Create a unique route handler for each Lambda function
    def create_route_handler(handler_func):
        def route_handler(*args, **kwargs):
            event = {
                "httpMethod": request.method,
                "path": request.path,
                "queryStringParameters": request.args.to_dict(),
                "headers": dict(request.headers),
                "body": request.get_data(as_text=True),
                "pathParameters": kwargs,
            }
            context = LambdaContext(resource_name)
            response = handler_func(event, context)

            try:
                api_response = APIResponse(**response)
                headers = response.get("headers", {})
                return Response(
                    api_response.body,
                    status=api_response.statusCode,
                    headers=headers,
                    mimetype="application/json",
                )
            except ValidationError as e:
                return jsonify({"error": "Invalid response format", "details": e.errors()}), 500

        return route_handler

    # Use a unique endpoint name for each route
    endpoint_name = f"{resource_name}_{method}_{path.replace('/', '_')}"
    app.add_url_rule(
        path,
        endpoint=endpoint_name,
        view_func=create_route_handler(handler_function),
        methods=[method.upper(), "OPTIONS"],
    )

登录后复制

你可以使用
启动你的服务器

if __name__ == "__main__":
    template = load_template()
    add_layers_to_path(template)
    setup_routes(template)
    app.run(debug=True, port=3000)

登录后复制

就是这样。完整代码可在 github https://github.com/JakubSzwajka/aws-sam-lambda-local-server-python 上找到。如果您发现任何带有图层等的极端情况，请告诉我。这可以改进，或者您认为值得添加更多内容。我觉得这很有帮助。