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-Lokalbefehle 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 eine Möglichkeit:

• 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 für jede Ä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 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)

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

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)

Flask-Routen erstellen

Im müssen wir Ressourcen durchlaufen und alle Funktionen finden. Auf dieser Grundlage erstelle ich einen 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

Dann besteht der nächste Schritt darin, es zu verwenden und für jede 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("{", "")

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

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)

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, dem 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?