”工欲善其事,必先利其器。“—孔子《论语.录灵公》
首页 > 编程 > AWS SAM Lambda 项目的本地开发服务器

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

发布于2024-11-02
浏览:446

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("{", "")

    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 上找到。如果您发现任何带有图层等的极端情况,请告诉我。这可以改进,或者您认为值得添加更多内容。我觉得这很有帮助。

潜在问题

简而言之,这适用于您的本地环境。请记住,lambda 具有一些内存限制和 cpu 限制。最后还是在真实环境下测试一下比较好。这种方法应该用于加速开发过程。

如果您在项目中实现了这一点,请分享您的见解。对你来说效果好吗?你遇到过什么挑战吗?您的反馈有助于为每个人改进此解决方案。

想了解更多吗?

请继续关注更多见解和教程!访问我的博客?

版本声明 本文转载于:https://dev.to/kuba_szw/local-development-server-for-aws-sam-lambda-projects-2dn2?1如有侵犯,请联系[email protected]删除
最新教程 更多>

免责声明: 提供的所有资源部分来自互联网,如果有侵犯您的版权或其他权益,请说明详细缘由并提供版权或权益证明然后发到邮箱:[email protected] 我们会第一时间内为您处理。

Copyright© 2022 湘ICP备2022001581号-3