AWS SAM Lambda プロジェクト用のローカル開発サーバー

現在、私は AWS ラムダをリクエスト ハンドラーとして使用して REST API を構築するプロジェクトに取り組んでいます。全体として AWS SAM を使用してラムダとレイヤーを定義し、それを優れた template.yaml ファイルで API ゲートウェイに接続します。

問題

この API をローカルでテストすることは、他のフレームワークほど簡単ではありません。 AWS は、ラムダ (Lambda 環境をより適切に複製する) をホストする Docker イメージを構築するための sam local コマンドを提供していますが、このアプローチは開発中に迅速に反復するには重すぎることがわかりました。

解決策

次のような方法が欲しかったです:

• ビジネス ロジックとデータ検証をすばやくテストします
• フロントエンド開発者がテストするためのローカル サーバーを提供します
• 変更のたびに Docker イメージを再構築するオーバーヘッドを回避します

そこで、これらのニーズに対応するスクリプトを作成しました。 ?‍♂️

TL;DR: この GitHub リポジトリのserver_local.py をチェックしてください。

主な利点

• クイックセットアップ: API Gateway ルートを Flask ルートにマッピングするローカル Flask サーバーを起動します。
• 直接実行: Docker のオーバーヘッドなしで、Python 関数 (Lambda ハンドラー) を直接トリガーします。
• ホットリロード: 変更はすぐに反映され、開発フィードバック ループが短縮されます。

この例は、sam init の「Hello World」プロジェクトに基づいて構築されており、ローカル開発を可能にするために server_local.py とその要件が追加されています。

SAM テンプレートの読み取り

ここで私がやっていることは、インフラストラクチャとすべてのラムダの現在の定義があるため、最初に template.yaml を読み取ることです。

辞書定義を作成するために必要なコードはこれだけです。 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 アプリではレイヤーが定義されていません。しかし、実際のプロジェクトではこの問題が発生しました。これを適切に動作させるために、レイヤー定義を読み取り、Python インポートが正しく動作できる sys.path に追加する関数を追加しました。これをチェックしてください:

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 で入手できます。レイヤーなどの例外的なケースを見つけた場合は、改善できる場合、またはこれにさらに何かを追加する価値があると思われる場合は、お知らせください。とても役に立ちます。

潜在的な問題

要するに、これはローカル環境で動作します。ラムダにはメモリと CPU にいくつかの制限が適用されることに注意してください。最終的には実際の環境でテストするのが良いでしょう。このアプローチは、開発プロセスをスピードアップするためにのみ使用する必要があります。

これをプロジェクトに実装する場合は、洞察を共有してください。うまくいきましたか？直面した課題はありますか?あなたのフィードバックは、すべての人にとってこのソリューションの改善に役立ちます。

もっと知りたいですか?

さらなる洞察とチュートリアルにご期待ください!私のブログにアクセスしてください?

以上がAWS SAM Lambda プロジェクト用のローカル開発サーバーの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。