No momento, estou trabalhando em um projeto em que a API REST é construída usando lambdas AWS como manipuladores de solicitação. A coisa toda usa AWS SAM para definir lambdas, camadas e conectá-los ao Api Gateway em um belo arquivo template.yaml.

O problema

Testar esta API localmente não é tão simples como com outras estruturas. Embora a AWS forneça comandos locais para construir imagens Docker que hospedam lambdas (que replicam melhor o ambiente Lambda), achei essa abordagem muito pesada para iterações rápidas durante o desenvolvimento.

A solução

Eu queria uma maneira de:

Teste rapidamente minha lógica de negócios e validações de dados
Forneça um servidor local para desenvolvedores front-end testarem
Evite a sobrecarga de reconstruir imagens do Docker para cada alteração

Então, criei um script para atender a essas necessidades. ?‍♂️

TL;DR: Confira server_local.py neste repositório GitHub.

Principais benefícios

• Configuração rápida: Aciona um servidor Flask local que mapeia suas rotas API Gateway para rotas Flask.
• Execução Direta: Aciona a função Python (manipulador Lambda) diretamente, sem sobrecarga do Docker.
• Hot Reload: As alterações são refletidas imediatamente, encurtando o ciclo de feedback do desenvolvimento.

Este exemplo baseia-se no projeto "Hello World" do sam init, com server_local.py e seus requisitos adicionados para permitir o desenvolvimento local.

Lendo o modelo SAM

O que estou fazendo aqui é ler o template.yaml primeiro, pois há uma definição atual de minha infraestrutura e de todos os lambdas.

Todo o código que precisamos para criar uma definição de dict é este. Para lidar com funções específicas do modelo SAM, adicionei alguns construtores ao CloudFormationLoader. Agora ele pode suportar Ref como referência a outro objeto, Sub como método para substituir e GetAtt para obter atributos. Acho que podemos adicionar mais lógica aqui, mas agora isso foi totalmente suficiente para fazer funcionar.

importar sistema operacional digitando import Any, Dict importar yaml classe CloudFormationLoader(yaml.SafeLoader): def __init__(self, fluxo): self._root = os.path.split(stream.name)[0] # tipo: ignorar super(CloudFormationLoader, self).__init__(stream) def include(self, nó): nome do arquivo = os.path.join(self._root, self.construct_scalar(node)) # tipo: ignorar com open(nome do arquivo, "r") como f: retornar yaml.load(f, CloudFormationLoader) def construct_getatt(carregador, nó): if isinstance(nó, yaml.ScalarNode): retornar {"Fn::GetAtt": loader.construct_scalar(node).split(".")} elif isinstance(nó, yaml.SequenceNode): retornar {"Fn::GetAtt": loader.construct_sequence(node)} outro: aumentar yaml.constructor.ConstructorError( Nenhum, Nenhum, f"Tipo de nó inesperado para! GetAtt: {type(node)}", node.start_mark ) CloudFormationLoader.add_constructor( "!Ref", carregador lambda, nó: {"Ref": loader.construct_scalar(node)} # tipo: ignorar ) CloudFormationLoader.add_constructor( "!Sub", carregador lambda, nó: {"Fn::Sub": loader.construct_scalar(node)} # tipo: ignorar ) CloudFormationLoader.add_constructor("!GetAtt",construct_getatt) def load_template() -> Dict[str, Qualquer]: com open("template.yaml", "r") como arquivo: retornar yaml.load(arquivo, Loader=CloudFormationLoader)

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)

E isso produz json assim:

{ "AWSTemplateFormatVersion":"09/09/2010", "Transformar":"AWS::Serverless-2016-10-31", "Description":"sam-app\nExemplo de modelo SAM para sam-app\n", "Globais":{ "Função":{ "Tempo limite":3, "Tamanho da memória": 128, "LoggingConfig":{ "LogFormat":"JSON" } } }, "Recursos":{ "HelloWorldFunction":{ "Tipo":"AWS::Serverless::Function", "Propriedades":{ "CodeUri":"hello_world/", "Handler":"app.lambda_handler", "Tempo de execução":"python3.9", "Arquiteturas":[ "x86_64" ], "Eventos":{ "Olá, Mundo":{ "Tipo":"API", "Propriedades":{ "Caminho":"/olá", "Método":"obter" } } } } } }, "Saídas":{ "HelloWorldApi":{ "Description":"URL do endpoint do API Gateway para estágio Prod para função Hello World", "Valor":{ "Fn::Sub":"https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/hello/" } }, "HelloWorldFunction":{ "Descrição":"Olá Mundo Função Lambda ARN", "Valor":{ "Fn::GetAtt":[ "HelloWorldFunction", "Arn" ] } }, "HelloWorldFunctionIamRole":{ "Description":"Papel IAM implícito criado para a função Hello World", "Valor":{ "Fn::GetAtt":[ "HelloWorldFunctionRole", "Arn" ] } } } }

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)

Manipulando Camadas

Sendo assim, é fácil criar rotas Flask dinamicamente para cada endpoint. Mas antes disso algo extra.

No aplicativo sam init helloworld não há camadas definidas. Mas eu tive esse problema no meu projeto real. Para que funcione corretamente, adicionei uma função que lê as definições de camadas e as adiciona ao sys.path para que as importações do python possam funcionar corretamente. Verifique isto:

def add_layers_to_path(modelo: Dict[str, Qualquer]): """Adicione camadas ao caminho. Lê o modelo e adiciona as camadas ao caminho para facilitar as importações.""" recursos = template.get("Recursos", {}) para _, recurso em resources.items(): if resource.get("Type") == "AWS::Serverless::LayerVersion": layer_path = recurso.get("Propriedades", {}).get("ContentUri") se caminho_camada: caminho_completo = os.path.join(os.getcwd(), caminho_da_camada) se full_path não estiver em sys.path: sys.path.append(caminho_completo)

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)

Criando Rotas Flask

Precisamos percorrer os recursos e encontrar todas as funções. Com base nisso, estou criando dados necessários para rotas de frascos.

def export_endpoints(modelo: Dict[str, Any]) -> Lista[Dict[str, str]]: pontos finais = [] recursos = template.get("Recursos", {}) para nome_do_recurso, recurso em resources.items(): if resource.get("Type") == "AWS::Serverless::Function": propriedades = recurso.get("Propriedades", {}) eventos = propriedades.get("Eventos", {}) para event_name, evento em events.items(): if event.get("Tipo") == "Api": api_props = event.get("Propriedades", {}) caminho = api_props.get("Caminho") método = api_props.get("Método") manipulador = propriedades.get("Manipulador") code_uri = propriedades.get("CodeUri") se caminho e método e manipulador e code_uri: pontos de extremidade.append( { "caminho": caminho, "método": método, "manipulador": manipulador, "código_uri": código_uri, "nome_do_recurso": nome_do_recurso, } ) pontos de extremidade de retorno

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)

O próximo passo é usá-lo e configurar uma rota para cada um.

def setup_routes(modelo: Dict[str, Any]): endpoints = export_endpoints(modelo) para endpoint em endpoints: rota_configuração( ponto final["caminho"], ponto final["método"], ponto final["manipulador"], ponto final["code_uri"], ponto final["nome_do_recurso"], ) def setup_route(caminho: str, método: str, manipulador: str, code_uri: str, nome_do_recurso: str): nome_do_módulo, nome_da_função = manipulador.rsplit(".", 1) module_path = os.path.join(code_uri, f"{module_name}.py") especificação = importlib.util.spec_from_file_location(module_name, module_path) se spec for None ou spec.loader for None: raise Exception(f"Módulo {module_name} não encontrado em {code_uri}") módulo = importlib.util.module_from_spec(especificação) spec.loader.exec_module(módulo) função_manipuladora = getattr(módulo, nome_função) caminho = caminho.replace("{", "") print(f"Configurando rota para [{method}] {path} com manipulador {resource_name}.") # Crie um manipulador de rota exclusivo para cada função Lambda def create_route_handler(handler_func): def route_handler(*args, **kwargs): evento = { "httpMethod": request.method, "caminho": solicitação.caminho, "queryStringParameters": request.args.to_dict(), "cabeçalhos": dict(request.headers), "corpo": request.get_data(as_text=True), "pathParameters": kwargs, } contexto = LambdaContext(nome_do_recurso) resposta = handler_func(evento, contexto) tentar: api_response = APIResponse(**resposta) cabeçalhos = resposta.get("cabeçalhos", {}) retornar Resposta ( api_response.body, status=api_response.statusCode, cabeçalhos=cabeçalhos, mimetype="aplicativo/json", ) exceto ValidationError como e: return jsonify({"error": "Formato de resposta inválido", "detalhes": e.errors()}), 500 retornar route_handler # Use um nome de endpoint exclusivo para cada rota endpoint_name = f"{nome_do_recurso}_{método}_{path.replace('/', '_')}" app.add_url_rule( caminho, ponto final=nome_ponto_final, view_func=create_route_handler(função_manipulador), métodos=[method.upper(), "OPÇÕES"], )

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)

E você pode iniciar seu servidor com

se __name__ == "__main__": modelo = carregar_modelo() add_layers_to_path(modelo) setup_routes(modelo) app.run(debug=True, porta=3000)

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)

É isso. Todo o código disponível no github https://github.com/JakubSzwajka/aws-sam-lambda-local-server-python. Deixe-me saber se você encontrar algum canto com camadas, etc. Isso pode ser melhorado ou você acha que vale a pena adicionar algo mais a isso. Acho isso muito útil.

Problemas potenciais

Resumindo, isso funciona no seu ambiente local. Tenha em mente que lambdas tem algumas limitações de memória e CPU aplicadas. No final é bom testá-lo em ambiente real. Esta abordagem deve ser usada apenas para acelerar o processo de desenvolvimento.

Se você implementar isso em seu projeto, compartilhe seus insights. Funcionou bem para você? Algum desafio que você enfrentou? Seu feedback ajuda a melhorar esta solução para todos.

Quer saber mais?

Fique ligado para mais insights e tutoriais! Visite meu blog?