En ce moment, je travaille sur un projet dans lequel l'API REST est construite en utilisant les lambdas AWS comme gestionnaires de requêtes. Le tout utilise AWS SAM pour définir des lambdas, des couches et le connecter à Api Gateway dans un joli fichier template.yaml.
Tester cette API localement n'est pas aussi simple qu'avec d'autres frameworks. Alors qu'AWS fournit des commandes locales pour créer des images Docker qui hébergent des lambdas (qui reproduisent mieux l'environnement Lambda), j'ai trouvé cette approche trop lourde pour des itérations rapides pendant le développement.
Je voulais un moyen de :
J'ai donc créé un script pour répondre à ces besoins. ?♂️
TL;DR : consultez server_local.py dans ce référentiel GitHub.
Cet exemple s'appuie sur le projet "Hello World" de sam init, avec server_local.py et ses exigences ajoutées pour permettre le développement local.
Ce que je fais ici, c'est que je lis d'abord le template.yaml car il y a la définition actuelle de mon infrastructure et de tous les lambdas.
Tout le code dont nous avons besoin pour créer une définition de dict est le suivant. Pour gérer les fonctions spécifiques au modèle SAM, j'ai ajouté quelques constructeurs à CloudFormationLoader. Il peut désormais prendre en charge Ref comme référence à un autre objet, Sub comme méthode de substitution et GetAtt pour obtenir des attributs. Je pense que nous pouvons ajouter plus de logique ici, mais pour le moment, c'était tout à fait suffisant pour que cela fonctionne.
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)
Et cela produit du json comme ceci :
{
"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"
]
}
}
}
}
Ainsi, il est facile de créer dynamiquement des itinéraires Flask pour chaque point de terminaison. Mais avant cela, quelque chose en plus.
Dans l'application sam init helloworld, aucun calque n'est défini. Mais j'ai eu ce problème dans mon vrai projet. Pour que cela fonctionne correctement, j'ai ajouté une fonction qui lit les définitions de couches et les ajoute à sys.path pour que les importations Python puissent fonctionner correctement. Vérifiez ceci :
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)
Dans le cas, nous devons parcourir les ressources et trouver toutes les fonctions. Sur cette base, je crée un besoin de données pour les itinéraires de flacons.
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
Ensuite, l'étape suivante consiste à l'utiliser et à configurer un itinéraire pour chacun.
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"],
)
Et vous pouvez démarrer votre serveur avec
if __name__ == "__main__":
template = load_template()
add_layers_to_path(template)
setup_routes(template)
app.run(debug=True, port=3000)
C'est tout. L'intégralité du code disponible sur github https://github.com/JakubSzwajka/aws-sam-lambda-local-server-python. Faites-moi savoir si vous trouvez un cas d'angle avec des calques, etc. Cela peut être amélioré ou vous pensez que cela vaut la peine d'ajouter quelque chose de plus à cela. Je trouve cela très utile.
En bref cela fonctionne sur votre environnement local. Gardez à l'esprit que lambdas a certaines limitations de mémoire appliquées et de processeur. Au final, c'est bien de le tester en environnement réel. Cette approche devrait être utilisée simplement pour accélérer le processus de développement.
Si vous mettez cela en œuvre dans votre projet, veuillez partager vos idées. Est-ce que ça a bien fonctionné pour vous ? Des défis que vous avez rencontrés ? Vos commentaires contribuent à améliorer cette solution pour tout le monde.
Restez à l’écoute pour plus d’informations et de tutoriels ! Visiter mon blog ?
Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!
Configurer les variables d'environnement Java
Qu'est-ce qu'un servomoteur
Comment enregistrer des fichiers sur une clé USB
écran de téléphone portable tft
L'ordinateur demande une solution nsiserror
Méthodes de réparation des vulnérabilités des bases de données
utilisation de la fonction heure locale
utilisation d'ostringstream
![[Web front-end] Démarrage rapide de Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
