지금 저는 AWS 람다를 요청 핸들러로 사용하여 REST API를 구축하는 프로젝트를 진행하고 있습니다. 모든 작업은 AWS SAM을 사용하여 람다, 레이어를 정의하고 이를 멋진 template.yaml 파일로 Api 게이트웨이에 연결합니다.
이 API를 로컬에서 테스트하는 것은 다른 프레임워크만큼 간단하지 않습니다. AWS는 람다를 호스팅하는 Docker 이미지(Lambda 환경을 더 잘 복제함)를 구축하기 위한 sam 로컬 명령을 제공하지만 개발 중에 빠른 반복을 수행하기에는 이 접근 방식이 너무 무겁다는 것을 알았습니다.
내가 원하는 방법은 다음과 같습니다.
그래서 저는 이러한 요구 사항을 해결하기 위해 스크립트를 만들었습니다. ?♂️
TL;DR: 이 GitHub 저장소에서 server_local.py를 확인하세요.
이 예는 sam init의 "Hello World" 프로젝트를 기반으로 하며 로컬 개발을 가능하게 하기 위해 server_local.py 및 해당 요구 사항이 추가되었습니다.
제가 여기서 하는 일은 인프라와 모든 람다에 대한 현재 정의가 있으므로 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("{", "<").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 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
