En este momento estoy trabajando en un proyecto en el que se crea una API REST utilizando AWS lambdas como controladores de solicitudes. Todo utiliza AWS SAM para definir lambdas, capas y conectarlo a Api Gateway en un bonito archivo template.yaml.

El problema

Probar esta API localmente no es tan sencillo como con otros marcos. Si bien AWS proporciona comandos locales de Sam para crear imágenes de Docker que alojan lambdas (que replican mejor el entorno Lambda), encontré que este enfoque es demasiado pesado para iteraciones rápidas durante el desarrollo.

La solución

Quería una forma de:

• Pruebe rápidamente mi lógica empresarial y validaciones de datos
• Proporcionar un servidor local para que los desarrolladores frontend lo prueben
• Evite la sobrecarga de reconstruir imágenes de Docker para cada cambio

Entonces, creé un script para abordar estas necesidades. ?‍♂️

TL;DR: consulte server_local.py en este repositorio de GitHub.

Beneficios clave

• Configuración rápida: activa un servidor Flask local que asigna las rutas de tu API Gateway a las rutas de Flask.
• Ejecución directa: activa la función Python (controlador Lambda) directamente, sin la sobrecarga de Docker.
• Recarga en caliente: los cambios se reflejan inmediatamente, lo que acorta el ciclo de retroalimentación del desarrollo.

Este ejemplo se basa en el proyecto "Hello World" de sam init, con server_local.py y sus requisitos agregados para permitir el desarrollo local.

Leyendo la plantilla SAM

Lo que estoy haciendo aquí es leer template.yaml primero, ya que existe una definición actual de mi infraestructura y todas las lambdas.

Todo el código que necesitamos para crear una definición de dict es este. Para manejar funciones específicas de la plantilla SAM, agregué algunos constructores a CloudFormationLoader. Ahora puede admitir Ref como referencia a otro objeto, Sub como método para sustituir y GetAtt para obtener atributos. Creo que podemos agregar más lógica aquí, pero por ahora esto fue totalmente suficiente para que funcione.

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)

Y esto produce un json como este:

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

Manejo de capas

Teniendo eso en cuenta, es fácil crear dinámicamente rutas Flask para cada punto final. Pero antes de eso, algo extra.

En la aplicación sam init helloworld no hay capas definidas. Pero tuve este problema en mi proyecto real. Para que funcione correctamente, agregué una función que lee las definiciones de capas y las agrega a sys.path para que las importaciones de Python puedan funcionar correctamente. Mira esto:

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)

Crear rutas de matraz

En el necesitamos recorrer los recursos y encontrar todas las funciones. En base a eso, estoy creando la necesidad de datos para rutas de matraces.

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

El siguiente paso es usarlo y configurar una ruta para cada uno.

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("{", "")

    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"],
    )

Y puedes iniciar tu servidor con

if __name__ == "__main__":
    template = load_template()
    add_layers_to_path(template)
    setup_routes(template)
    app.run(debug=True, port=3000)

Eso es todo. El código completo disponible en github https://github.com/JakubSzwajka/aws-sam-lambda-local-server-python. Avíseme si encuentra algún caso de esquina con capas, etc. Eso se puede mejorar o cree que vale la pena agregar algo más a esto. Lo encuentro muy útil.

Problemas potenciales

En resumen, esto funciona en su entorno local. Tenga en cuenta que lambdas tiene aplicadas algunas limitaciones de memoria y CPU. Al final es bueno probarlo en un entorno real. Este enfoque debería utilizarse simplemente para acelerar el proceso de desarrollo.

Si implementas esto en tu proyecto, comparte tus ideas. ¿Te funcionó bien? ¿Algún desafío que haya enfrentado? Sus comentarios ayudan a mejorar esta solución para todos.

¿Quieres saber más?

¡Estén atentos para obtener más información y tutoriales! ¿Visitar mi blog?