現在我正在開發一個項目，其中使用 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 上找到。如果您發現任何帶有圖層等的極端情況，請告訴我。這可以改進，或者您認為值得添加更多內容。我覺得這很有幫助。
潛在問題
簡而言之，這適用於您當地的環境。請記住，lambda 具有一些記憶體限制和 cpu 限制。最後還是在真實環境下測試一下比較好。這種方法應該用來加速開發過程。
如果您在專案中實現了這一點，請分享您的見解。對你來說效果好嗎？你遇過什麼挑戰嗎？您的回饋有助於為每個人改進此解決方案。
想了解更多嗎？
請繼續關注更多見解和教學！訪問我的部落格？
以上是AWS SAM Lambda 專案的本機開發伺服器的詳細內容。更多資訊請關注PHP中文網其他相關文章！
