Im Moment arbeite ich an einem Projekt, bei dem die REST-API unter Verwendung von AWS-Lambdas als Anforderungshandler erstellt wird. Das Ganze verwendet AWS SAM, um Lambdas und Ebenen zu definieren und es in einer schönen template.yaml-Datei mit Api Gateway zu verbinden.
Das Problem
Das lokale Testen dieser API ist nicht so einfach wie bei anderen Frameworks. Während AWS Sam-Local-Befehle bereitstellt, um Docker-Images zu erstellen, die Lambdas hosten (die die Lambda-Umgebung besser replizieren), fand ich diesen Ansatz zu schwerfällig für schnelle Iterationen während der Entwicklung.
Die Lösung
Ich wollte einen Weg:
- Testen Sie schnell meine Geschäftslogik und Datenvalidierungen
- Stellen Sie einen lokalen Server für Frontend-Entwickler zum Testen bereit
- Vermeiden Sie den Aufwand, Docker-Images bei jeder Änderung neu zu erstellen
Also habe ich ein Skript erstellt, um diesen Anforderungen gerecht zu werden. ?♂️
TL;DR: Schauen Sie sich server_local.py in diesem GitHub-Repository an.
Hauptvorteile
-
Schnelleinrichtung: Richtet einen lokalen Flask-Server ein, der Ihre API-Gateway-Routen Flask-Routen zuordnet.
-
Direkte Ausführung: Löst die Python-Funktion (Lambda-Handler) direkt aus, ohne Docker-Overhead.
-
Hot Reload: Änderungen werden sofort übernommen, wodurch die Entwicklungs-Feedbackschleife verkürzt wird.
Dieses Beispiel baut auf dem „Hello World“-Projekt von sam init auf, wobei server_local.py und seine Anforderungen hinzugefügt wurden, um die lokale Entwicklung zu ermöglichen.
Lesen der SAM-Vorlage
Was ich hier mache, ist, dass ich zuerst die template.yaml lese, da es eine aktuelle Definition meiner Infrastruktur und aller Lambdas gibt.
Der gesamte Code, den wir zum Erstellen einer Diktatdefinition benötigen, ist dieser. Um spezifische Funktionen für die SAM-Vorlage zu verarbeiten, habe ich CloudFormationLoader einige Konstruktoren hinzugefügt. Es kann jetzt Ref als Referenz auf ein anderes Objekt, Sub als Methode zum Ersetzen und GetAtt zum Abrufen von Attributen unterstützen. Ich denke, wir können hier noch mehr Logik hinzufügen, aber im Moment war das völlig ausreichend, damit es funktioniert.
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)
Nach dem Login kopieren
Und das erzeugt JSON wie folgt:
{
"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"
]
}
}
}
}
Nach dem Login kopieren
Umgang mit Ebenen
Dadurch ist es einfach, Flask-Routen für jeden Endpunkt dynamisch zu erstellen. Aber vorher noch etwas Besonderes.
In der Sam-Init-Helloworld-App sind keine Ebenen definiert. Aber ich hatte dieses Problem in meinem realen Projekt. Damit es ordnungsgemäß funktioniert, habe ich eine Funktion hinzugefügt, die Ebenendefinitionen liest und sie zu sys.path hinzufügt, damit Python-Importe ordnungsgemäß funktionieren. Überprüfen Sie Folgendes:
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)
Nach dem Login kopieren
Flask-Routen erstellen
Im müssen wir Ressourcen durchgehen und alle Funktionen finden. Auf dieser Grundlage erstelle ich den Datenbedarf für Flaschenrouten.
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
Nach dem Login kopieren
Dann besteht der nächste Schritt darin, es zu verwenden und für jeden eine Route einzurichten.
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"],
)
Nach dem Login kopieren
Und Sie können Ihren Server starten mit
if __name__ == "__main__":
template = load_template()
add_layers_to_path(template)
setup_routes(template)
app.run(debug=True, port=3000)
Nach dem Login kopieren
Das ist es. Der gesamte Code ist auf Github verfügbar: https://github.com/JakubSzwajka/aws-sam-lambda-local-server-python. Lassen Sie mich wissen, wenn Sie ein Eckgehäuse mit Schichten usw. finden. Das kann verbessert werden, oder Sie denken, es lohnt sich, etwas mehr hinzuzufügen. Ich finde es sehr hilfreich.
Mögliche Probleme
Kurz gesagt, dies funktioniert in Ihrer lokalen Umgebung. Beachten Sie, dass bei Lambdas einige Speicher- und CPU-Einschränkungen gelten. Am Ende ist es gut, es in einer realen Umgebung zu testen. Dieser Ansatz sollte lediglich verwendet werden, um den Entwicklungsprozess zu beschleunigen.
Wenn Sie dies in Ihrem Projekt umsetzen, teilen Sie uns bitte Ihre Erkenntnisse mit. Hat es bei Ihnen gut funktioniert? Gab es irgendwelche Herausforderungen, mit denen Sie konfrontiert waren? Ihr Feedback trägt dazu bei, diese Lösung für alle zu verbessern.
Möchten Sie mehr wissen?
Bleiben Sie dran für weitere Einblicke und Tutorials! Besuchen Sie meinen Blog ?
Das obige ist der detaillierte Inhalt vonLokaler Entwicklungsserver für AWS SAM Lambda-Projekte. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
