AWS SAM Lambda 專案的本機開發伺服器-Python教學-PHP中文網

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 上找到。如果您發現任何帶有圖層等的極端情況，請告訴我。這可以改進，或者您認為值得添加更多內容。我覺得這很有幫助。