지금 저는 AWS 람다를 요청 핸들러로 사용하여 REST API를 구축하는 프로젝트를 진행하고 있습니다. 모든 것은 AWS SAM을 사용하여 람다, 레이어를 정의하고 이를 멋진 template.yaml 파일의 Api Gateway에 연결합니다.

문제

이 API를 로컬에서 테스트하는 것은 다른 프레임워크만큼 간단하지 않습니다. AWS는 람다를 호스팅하는 Docker 이미지(Lambda 환경을 더 잘 복제함)를 구축하기 위한 sam 로컬 명령을 제공하지만 개발 중에 빠른 반복을 수행하기에는 이 접근 방식이 너무 무겁다는 것을 알았습니다.

해결책

나는 다음과 같은 방법을 원했습니다:

• 비즈니스 로직 및 데이터 유효성 검사를 빠르게 테스트
• 프런트엔드 개발자가 테스트할 수 있는 로컬 서버 제공
• 변경할 때마다 Docker 이미지를 다시 빌드하는 오버헤드 방지

그래서 저는 이러한 요구 사항을 해결하기 위해 스크립트를 만들었습니다. ?‍♂️

TL;DR: 이 GitHub 저장소에서 server_local.py를 확인하세요.

주요 이점

• 빠른 설정: API 게이트웨이 경로를 Flask 경로에 매핑하는 로컬 Flask 서버를 가동합니다.
• 직접 실행: Docker 오버헤드 없이 Python 함수(Lambda 핸들러)를 직접 트리거합니다.
• 핫 리로드: 변경 사항이 즉시 반영되어 개발 피드백 루프가 단축됩니다.

이 예제는 sam init의 "Hello World" 프로젝트를 기반으로 하며 로컬 개발을 가능하게 하기 위해 server_local.py 및 해당 요구 사항이 추가되었습니다.

SAM 템플릿 읽기

제가 여기서 하고 있는 일은 인프라와 모든 람다에 대한 현재 정의가 있으므로 template.yaml을 먼저 읽는 것입니다.

dict 정의를 생성하는 데 필요한 모든 코드는 다음과 같습니다. 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)

플라스크 경로 생성

리소스 전체를 반복하고 모든 기능을 찾아야 합니다. 이를 바탕으로 플라스크 루트에 필요한 데이터를 작성하고 있습니다.

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에서 확인할 수 있습니다. 레이어 등이 포함된 코너 케이스를 찾으면 알려주세요. 개선할 수 있거나 여기에 뭔가를 더 추가할 가치가 있다고 생각합니다. 나는 그것이 매우 도움이 된다고 생각한다.

잠재적인 문제

간단히 말하면 이는 로컬 환경에서 작동합니다. 람다에는 일부 메모리 제한이 적용되고 CPU가 있다는 점을 명심하세요. 결국 실제 환경에서 테스트하는 것이 좋습니다. 이 접근 방식은 개발 프로세스 속도를 높이는 데 사용해야 합니다.

프로젝트에 이 기능을 구현한다면 통찰력을 공유해 주세요. 효과가 좋았나요? 직면한 어려움이 있나요? 귀하의 의견은 모두를 위해 이 솔루션을 개선하는 데 도움이 됩니다.

더 알고 싶으십니까?

더 많은 통찰력과 튜토리얼을 기대해주세요! 내 블로그를 방문하시겠습니까?