Сейчас я работаю над проектом, в котором REST API построен с использованием лямбда-выражений AWS в качестве обработчиков запросов. Все это использует AWS SAM для определения лямбда-выражений, слоев и подключения их к шлюзу Api в красивом файле template.yaml.

Проблема

Локальное тестирование этого API не так просто, как с другими платформами. Хотя AWS предоставляет одинаковые локальные команды для создания образов Docker, содержащих лямбда-выражения (которые лучше копируют среду Lambda), я нашел этот подход слишком тяжелым для быстрых итераций во время разработки.

Решение

Мне нужен был способ:

• Быстро протестируйте мою бизнес-логику и проверку данных
• Предоставить локальный сервер разработчикам внешнего интерфейса для тестирования
• Избегайте накладных расходов на пересборку образов Docker для каждого изменения

Итак, я создал сценарий для удовлетворения этих потребностей. ?‍♂️

TL;DR: посмотрите server_local.py в этом репозитории GitHub.

Ключевые преимущества

• Быстрая настройка: запускает локальный сервер Flask, который сопоставляет маршруты шлюза API с маршрутами Flask.
• Прямое выполнение: запускает функцию Python (обработчик Lambda) напрямую, без дополнительных затрат Docker.
• Горячая перезагрузка: изменения отражаются немедленно, что сокращает цикл обратной связи при разработке.

Этот пример основан на проекте «Hello World» из sam init, с добавлением server_local.py и его требований для обеспечения возможности локальной разработки.

Чтение шаблона SAM

Здесь я сначала читаю template.yaml, так как там есть текущее определение моей инфраструктуры и все лямбда-выражения.

Весь код, который нам нужен для создания определения dict, следующий. Для обработки функций, специфичных для шаблона SAM, я добавил в CloudFormationLoader несколько конструкторов. Теперь он может поддерживать Ref как ссылку на другой объект, Sub как метод для замены и GetAtt для получения атрибутов. Я думаю, мы можем добавить сюда больше логики, но сейчас этого было вполне достаточно, чтобы всё заработало.

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)

И это создает такой JSON:

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

Обработка слоев

При этом можно легко динамически создавать маршруты Flask для каждой конечной точки. Но перед этим кое-что дополнительное.

В приложении sam init helloworld не определены слои. Но у меня была эта проблема в моем реальном проекте. Чтобы все работало правильно, я добавил функцию, которая считывает определения слоев и добавляет их в sys.path, чтобы импорт Python мог работать правильно. Проверьте это:

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-маршрутов

В цикле нам нужно просмотреть ресурсы и найти все функции. На основании этого я создаю данные, необходимые для маршрутов колб.

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

Затем следующий шаг — использовать его и настроить маршрут для каждого из них.

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

И вы можете запустить свой сервер с помощью

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

Вот и все. Весь код доступен на github https://github.com/JakubSzwajka/aws-sam-lambda-local-server-python. Дайте мне знать, если вы найдете какой-либо угловой случай со слоями и т. д. Это можно улучшить или вы думаете, что стоит добавить к этому что-то еще. Я считаю это очень полезным.

Потенциальные проблемы

Короче говоря, это работает в вашей локальной среде. Имейте в виду, что лямбда-выражения имеют некоторые ограничения по памяти и процессору. В конце концов, полезно протестировать это в реальной среде. Этот подход следует использовать только для ускорения процесса разработки.

Если вы реализуете это в своем проекте, поделитесь своим мнением. Это сработало для вас? Какие-нибудь проблемы, с которыми вы столкнулись? Ваш отзыв поможет улучшить это решение для всех.

Хотите узнать больше?

Следите за новостями и обучающими материалами! Посетите мой блог ?