Sekarang saya sedang mengusahakan projek di mana REST API dibina menggunakan AWS lambdas sebagai pengendali permintaan. Keseluruhannya menggunakan AWS SAM untuk mentakrifkan lambda, lapisan dan menyambungkannya ke Api Gateway dalam fail template.yaml yang bagus.
Masalahnya
Menguji API ini secara setempat tidak semudah dengan rangka kerja lain. Walaupun AWS menyediakan arahan setempat yang sama untuk membina imej Docker yang mengehoskan lambdas (yang lebih baik meniru persekitaran Lambda), saya mendapati pendekatan ini terlalu berat untuk lelaran pantas semasa pembangunan.
Penyelesaian
Saya mahukan cara untuk:
- Uji logik perniagaan dan pengesahan data saya dengan cepat
- Sediakan pelayan tempatan untuk diuji oleh pembangun bahagian hadapan
- Elakkan overhed membina semula imej Docker untuk setiap perubahan
Jadi, saya mencipta skrip untuk memenuhi keperluan ini. ?♂️
TL;DR: Lihat server_local.py dalam repositori GitHub ini.
Faedah Utama
-
Persediaan Pantas: Memusingkan pelayan Flask setempat yang memetakan laluan API Gateway anda ke laluan Flask.
-
Pelaksanaan Terus: Mencetuskan fungsi Python (pengendali Lambda) secara langsung, tanpa overhed Docker.
-
Muat Semula Panas: Perubahan ditunjukkan serta-merta, memendekkan gelung maklum balas pembangunan.
Contoh ini dibina pada projek "Hello World" daripada sam init, dengan server_local.py dan keperluannya ditambah untuk membolehkan pembangunan setempat.
Membaca Templat SAM
Apa yang saya lakukan di sini ialah saya membaca templat.yaml dahulu kerana terdapat takrif semasa infrastruktur saya dan semua lambda.
Semua kod yang kita perlukan untuk mencipta definisi dict adalah ini. Untuk mengendalikan fungsi khusus untuk templat SAM, saya telah menambah beberapa pembina pada CloudFormationLoader. Ia kini boleh menyokong Ref sebagai rujukan kepada objek lain, Sub sebagai kaedah untuk menggantikan dan GetAtt untuk mendapatkan atribut. Saya rasa kita boleh menambah lebih logik di sini tetapi sekarang ini sudah memadai untuk menjadikannya berfungsi.
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)
Salin selepas log masuk
Dan ini menghasilkan json seperti ini:
{
"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"
]
}
}
}
}
Salin selepas log masuk
Mengendalikan Lapisan
Melalui itu adalah mudah untuk mencipta laluan Flask secara dinamik untuk setiap titik akhir. Tetapi sebelum itu sesuatu yang lebih.
Dalam apl sam init helloworld tiada lapisan yang ditentukan. Tetapi saya mempunyai masalah ini dalam projek sebenar saya. Untuk menjadikannya berfungsi dengan betul, saya telah menambahkan fungsi yang membaca definisi lapisan dan menambahkannya pada sys.path yang import python boleh berfungsi dengan betul. Semak ini:
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)
Salin selepas log masuk
Mencipta Laluan Flask
Dalam kita perlu mengulangi seluruh sumber dan mencari semua fungsi. Berdasarkan itu, saya mencipta keperluan data untuk laluan kelalang.
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
Salin selepas log masuk
Kemudian langkah seterusnya ialah menggunakannya dan menyediakan laluan untuk setiap satu.
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"],
)
Salin selepas log masuk
Dan anda boleh memulakan pelayan anda dengan
if __name__ == "__main__":
template = load_template()
add_layers_to_path(template)
setup_routes(template)
app.run(debug=True, port=3000)
Salin selepas log masuk
Itu sahaja. Keseluruhan kod tersedia di github https://github.com/JakubSzwajka/aws-sam-lambda-local-server-python. Beritahu saya jika anda menjumpai mana-mana bekas sudut dengan lapisan dan lain-lain. Itu boleh dipertingkatkan atau anda fikir ia berbaloi untuk menambah sesuatu yang lebih pada ini. Saya dapati ia sangat membantu.
Isu Berpotensi
Ringkasnya, ini berfungsi pada persekitaran tempatan anda. Perlu diingat bahawa lambdas mempunyai beberapa had memori yang digunakan dan cpu. Pada akhirnya adalah baik untuk mengujinya dalam persekitaran sebenar. Pendekatan ini harus digunakan untuk mempercepatkan proses pembangunan.
Jika anda melaksanakan perkara ini dalam projek anda, sila kongsikan pandangan anda. Adakah ia berfungsi dengan baik untuk anda? Sebarang cabaran yang anda hadapi? Maklum balas anda membantu memperbaik penyelesaian ini untuk semua orang.
Ingin Tahu Lebih Lanjut?
Nantikan lebih banyak cerapan dan tutorial! Lawati Blog Saya ?
Atas ialah kandungan terperinci Pelayan Pembangunan Tempatan untuk Projek Lambda AWS SAM. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!
![[Web front-end] Permulaan pantas Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
