Serveur de développement local pour les projets AWS SAM Lambda

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.

Le problème

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.

La solution

Je voulais un moyen de :

• Tester rapidement ma logique métier et mes validations de données
• Fournir un serveur local permettant aux développeurs frontend de tester
• Évitez les frais liés à la reconstruction des images Docker pour chaque modification

J'ai donc créé un script pour répondre à ces besoins. ?‍♂️

TL;DR : consultez server_local.py dans ce référentiel GitHub.

Avantages clés

• Configuration rapide : démarre un serveur Flask local qui mappe vos routes API Gateway aux routes Flask.
• Exécution directe : Déclenche la fonction Python (gestionnaire Lambda) directement, sans surcharge Docker.
• Rechargement à chaud : les modifications sont reflétées immédiatement, raccourcissant ainsi la boucle de rétroaction sur le développement.

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.

Lecture du modèle SAM

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"
            ]
         }
      }
   }
}

Gestion des calques

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)

Création d'itinéraires de flacons

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.

Problèmes potentiels

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.

Vous voulez en savoir plus ?

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!