Estruture um projeto Python em arquitetura hexagonal usando o AWS Lambda - Recomendações da AWS
ResumoPré-requisitos e limitaçõesArquiteturaFerramentasPráticas recomendadasÉpicosRecursos relacionados

As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Estruture um projeto Python em arquitetura hexagonal usando o AWS Lambda

Furkan Oruc, Dominik Goby, Darius Kunce e Michal Ploski, da Amazon Web Services

Resumo

Esse padrão mostra como estruturar um projeto Python em arquitetura hexagonal usando o AWS Lambda. O padrão usa o AWS Cloud Development Kit (AWS CDK) como ferramenta de infraestrutura como código (IaC), o Amazon API Gateway como API REST e o Amazon DynamoDB como camada de persistência. A arquitetura hexagonal segue os princípios de design orientados por domínio. Na arquitetura hexagonal, o software consiste em três componentes: domínio, portas e adaptadores. Para obter informações detalhadas sobre arquiteturas hexagonais e seus benefícios, consulte o guia Criação de arquiteturas hexagonais na AWS.

Pré-requisitos e limitações

Pré-requisitos

Versões do produto

  • Git versão 2.24.3 ou superior

  • Python versão 3.7 ou superior

  • AWS CDK v2

  • Poetry versão 1.1.13 ou superior

  • AWS Lambda Powertools para Python versão 1.25.6 ou superior

  • pytest versão 7.1.1 ou superior

  • Moto versão 3.1.9 ou superior

  • pydantic versão 1.9.0 ou superior

  • Boto3 versão 1.22.4 ou superior

  • mypy-boto3-dynamodb versão 1.24.0 ou superior

Arquitetura

Pilha de tecnologias de destino

A pilha de tecnologia de destino consiste em um serviço em Python que usa o API Gateway, Lambda e DynamoDB. O serviço usa um adaptador do DynamoDB para manter os dados. Ele fornece uma função que usa o Lambda como ponto de entrada. O serviço usa o Amazon API Gateway para expor uma API REST. A API usa o AWS Identity and Access Management (IAM) para a autenticação de clientes.

Arquitetura de destino

Para ilustrar a implementação, esse padrão implanta uma arquitetura de destino com tecnologia sem servidor. Os clientes podem enviar solicitações para um endpoint do API Gateway. O API Gateway encaminha a solicitação para a função do Lambda de destino que implementa o padrão de arquitetura hexagonal. A função do Lambda executa operações de criação, leitura, atualização e exclusão (CRUD) em uma tabela do DynamoDB.

Importante

Esse padrão foi testado em um ambiente PoC. Você deve realizar uma análise de segurança para identificar o modelo de ameaça e criar uma base de código segura antes de implantar qualquer arquitetura em um ambiente de produção.

Arquitetura de destino para estruturar um projeto Python em arquitetura hexagonal

A API oferece suporte a cinco operações em uma entidade de produto:

  • GET /products devolve todos os produtos.

  • POST /products cria um novo produto.

  • GET /products/{id} retorna um produto específico.

  • PUT /products/{id} atualiza um produto específico.

  • DELETE /products/{id} exclui um produto específico.

Você pode usar a seguinte estrutura de pastas para organizar seu projeto de acordo com o padrão de arquitetura hexagonal:  

app/  # application code
|--- adapters/  # implementation of the ports defined in the domain
     |--- tests/  # adapter unit tests
|--- entrypoints/  # primary adapters, entry points
     |--- api/  # api entry point
          |--- model/  # api model
          |--- tests/  # end to end api tests
|--- domain/  # domain to implement business logic using hexagonal architecture
     |--- command_handlers/  # handlers used to execute commands on the domain
     |--- commands/  # commands on the domain
     |--- events/  # events triggered via the domain
     |--- exceptions/  # exceptions defined on the domain
     |--- model/  # domain model
     |--- ports/  # abstractions used for external communication
     |--- tests/  # domain tests
|--- libraries/  # List of 3rd party libraries used by the Lambda function
infra/  # infrastructure code
simple-crud-app.py  # AWS CDK v2 app

Ferramentas

Serviços da AWS

  • O Amazon API Gateway é um serviço totalmente gerenciado que facilita aos desenvolvedores criar, publicar, manter, monitorar e proteger APIs em qualquer escala.

  • O Amazon DynamoDB é um banco de dados NoSQL totalmente gerenciado, de valor-chave e com tecnologia sem servidor, projetado para executar aplicativos de alto desempenho em qualquer escala.

  • O AWS Lambda é um serviço computacional com tecnologia sem servidor e orientado a eventos que permite executar o código em virtualmente qualquer tipo de aplicação ou serviço de backend sem o provisionamento ou gerenciamento de servidores. Você pode iniciar funções do Lambda a partir de mais de 200 serviços da AWS e aplicativos de software como serviço (SaaS) e pagar somente pelo que usar.

Ferramentas

  • O Git  é usado como sistema de controle de versão para desenvolvimento de código nesse padrão.

  • O Python é usado como linguagem de programação para esse padrão. O Python fornece estruturas de dados de alto nível e uma abordagem à programação orientada a objetos. O AWS Lambda fornece um runtime integrado do Python que simplifica a operação dos serviços do Python.

  • O Visual Studio Code é usado como IDE para desenvolvimento e teste desse padrão. Você pode usar qualquer IDE que ofereça suporte ao desenvolvimento em Python (por exemplo, PyCharm).

  • O AWS Cloud Development Kit (AWS CDK) é uma estrutura de desenvolvimento de software de código aberto que permite definir recursos de aplicações em nuvem usando linguagens de programação conhecidas. Esse padrão usa o CDK para escrever e implantar a infraestrutura de nuvem como código.

  • O Poetry é usado para gerenciar dependências no padrão.

  • O Docker é usado pelo AWS CDK para criar o pacote e a camada do Lambda.

Código

O código desse padrão está disponível no repositório de amostras da arquitetura hexagonal GitHub Lambda.

Práticas recomendadas

Para usar esse padrão em um ambiente de produção, siga essas práticas recomendadas:

Esse padrão usa o AWS X-Ray para rastrear solicitações por meio do ponto de entrada, domínio e adaptadores do aplicativo. O AWS X-Ray ajuda os desenvolvedores a identificar gargalos e determinar altas latências para melhorar o desempenho do aplicativo.

Épicos

TarefaDescriçãoHabilidades necessárias

Crie seu próprio repositório.

  1. Faça login em GitHub.

  2. Crie um novo repositório. Para obter instruções, consulte a GitHub documentação.

  3. Clone e envie o repositório de amostra desse padrão para o novo repositório em sua conta.

Desenvolvedor de aplicativos

Instale as dependências.

  1. Instale o Poetry.

    pip install poetry

  2. Instale pacotes do diretório raiz. O comando a seguir instala o aplicativo e os pacotes do AWS CDK. Ele também instala pacotes de desenvolvimento necessários para a execução de testes de unidade. Todos os pacotes instalados são colocados em um novo ambiente virtual.

    poetry install

  3. Para visualizar uma representação gráfica dos pacotes instalados, execute o comando a seguir.

    poetry show --tree

  4. Atualizar todas as dependências.

    poetry update

  5. Abra um novo shell no ambiente virtual recém-criado. Ele contém todas as dependências instaladas.

    poetry shell
Desenvolvedor de aplicativos

Configure seu IDE.

Recomendamos o Visual Studio Code, mas você pode usar qualquer IDE de sua escolha que ofereça suporte ao Python. As etapas a seguir são para o Visual Studio Code.

  1. Atualize o arquivo .vscode/settings.

    {
    "python.testing.pytestArgs": [
        "app/adapters/tests",
        "app/entrypoints/api/tests",
        "app/domain/tests"
    ],
    "python.testing.unittestEnabled": false,
    "python.testing.pytestEnabled": true,
    "python.envFile": "${workspaceFolder}/.env",
}

  2. Crie um arquivo .env no diretório raiz do projeto. Isso garante que o diretório raiz do projeto seja incluído no PYTHONPATH para que pytest possa localizá-lo e descobrir todos os pacotes adequadamente.

    PYTHONPATH=.
Desenvolvedor de aplicativos

Execute testes de unidade, opção 1: usando o Visual Studio Code.

  1. Selecione o interpretador Python do ambiente virtual gerenciado pelo Poetry.

  2. Execute testes no Test Explorer.

Desenvolvedor de aplicativos

Execute testes de unidade, opção 2: usando comandos shell.

  1. Inicie um novo shell no ambiente virtual.

    poetry shell

  2. Execute o comando pytest no diretório raiz.

    python -m pytest

    Alternativamente, você pode executar o comando diretamente do Poetry.

    poetry run python -m pytest
Desenvolvedor de aplicativos
TarefaDescriçãoHabilidades necessárias

Solicite credenciais temporárias.

Para obter credenciais da AWS no shell durante a execução do cdk deploy, crie credenciais temporárias usando o Centro de Identidade do AWS IAM (sucessor do AWS Single Sign-On). Para obter instruções, consulte a publicação Como recuperar credenciais de curto prazo para uso da CLI com o Centro de Identidade do AWS IAM.

Desenvolvedor de aplicativos, AWS DevOps

Implante o aplicativo .

  1. Instale o AWS CDK v2.

    npm install -g aws-cdk

    Para obter mais informações, consulte a documentação do AWS CDK.

  2. Faça o bootstrap do AWS CDK na sua conta e região.

    cdk bootstrap aws://12345678900/us-east-1 --profile aws-profile-name

  3. Implante o aplicativo como uma CloudFormation pilha da AWS usando um perfil da AWS.

    cdk deploy --profile aws-profile-name
Desenvolvedor de aplicativos, AWS DevOps

Teste a API, opção 1: usando o console.

Use o console do API Gateway para testar a API. Para obter mais informações sobre as operações e request/response mensagens da API, consulte a seção Uso da API do arquivo readme no GitHub repositório.

Desenvolvedor de aplicativos, AWS DevOps

Teste a API, opção 2: usando o Postman.

Se você quiser usar uma ferramenta como o Postman:

  1. Instale o Postman como um aplicativo independente ou extensão do navegador.

  2. Copie o URL do endpoint para o API Gateway. Ele estará no seguinte formato.

    https://{api-id}.execute-api.{region}.amazonaws.com/{stage}/{path}

  3. Configure a assinatura da AWS na guia de autorização. Para obter instruções, consulte o artigo do AWS re:POST sobre a ativação da autenticação do IAM para o API Gateway REST. APIs

  4. Use o Postman para enviar solicitações ao endpoint da API.

Desenvolvedor de aplicativos, AWS DevOps
TarefaDescriçãoHabilidades necessárias

Escreva testes de unidade para o domínio comercial.

  1. Crie um arquivo Python na pasta app/domain/tests usando o prefixo do nome do arquivo test_.

  2. Crie um novo método de teste para testar a nova lógica de negócios usando o exemplo a seguir.

    def test_create_product_should_store_in_repository():
    # Arrange
    command = create_product_command.CreateProductCommand(
        name="Test Product",
        description="Test Description",
    )
    # Act
    create_product_command_handler.handle_create_product_command(
        command=command, unit_of_work=mock_unit_of_work
    )
 # Assert

  3. Crie uma classe de comando na pasta app/domain/commands. 

  4. Caso seja uma funcionalidade nova, crie um stub para o manipulador de comandos na pasta app/domain/command_handlers.

  5. Execute o teste de unidade para verificar se ela falha, porque ainda não há lógica de negócios.

    python -m pytest
Desenvolvedor de aplicativos

Implemente comandos e manipuladores de comandos.

  1. Implemente a lógica de negócios no arquivo manipulador de comandos recém-criado. 

  2. Para cada dependência que interage com sistemas externos, declare uma classe abstrata na pasta app/domain/ports.

    class ProductsRepository(ABC):
    @abstractmethod
    def add(self, product: product.Product) -> None:
        ...
 
class UnitOfWork(ABC):
    products: ProductsRepository

    @abstractmethod
    def commit(self) -> None:
        ...

    @abstractmethod
    def __enter__(self) -> typing.Any:
        ...

    @abstractmethod
    def __exit__(self, *args) -> None:
        ...

  3. Atualize a assinatura do manipulador de comandos para aceitar as dependências recém-declaradas usando a classe de porta abstrata como anotação de tipo.

    def handle_create_product_command(
    command: create_product_command.CreateProductCommand,
    unit_of_work: unit_of_work.UnitOfWork,
) -> str:
    ...

  4. Atualize o teste de unidade para simular o comportamento de todas as dependências declaradas para o manipulador de comandos.

        # Arrange
    mock_unit_of_work = unittest.mock.create_autospec(
        spec=unit_of_work.UnitOfWork, instance=True
    )
    mock_unit_of_work.products = unittest.mock.create_autospec(
        spec=unit_of_work.ProductsRepository, instance=True
    )

  5. Atualize a lógica de asserção no teste para verificar as invocações de dependência esperadas.

      # Assert
    mock_unit_of_work.commit.assert_called_once()
    product = mock_unit_of_work.products.add.call_args.args[0]

    assertpy.assert_that(product.name).is_equal_to("Test Product")
    assertpy.assert_that(product.description).is_equal_to("Test Description")

  6. Execute o teste de unidade para verificar o sucesso.

    python -m pytest
Desenvolvedor de aplicativos

Escreva testes de integração para adaptadores secundários.

  1. Crie um arquivo de teste na pasta app/adapters/tests usando test_ como prefixo do nome do arquivo.

  2. Use a biblioteca Moto para simular os serviços da AWS.

    @pytest.fixture 
def mock_dynamodb(): 
   with moto.mock_dynamodb(): 
    yield boto3.resource("dynamodb", region_name="eu-central-1")

  3. Crie um novo método de teste para um teste de integração do adaptador.

    def test_add_and_commit_should_store_product(mock_dynamodb):
    # Arrange
    unit_of_work = dynamodb_unit_of_work.DynamoDBUnitOfWork(
        table_name=TEST_TABLE_NAME, dynamodb_client=mock_dynamodb.meta.client
    )
    current_time = datetime.datetime.now(datetime.timezone.utc).isoformat()

    new_product_id = str(uuid.uuid4())
    new_product = product.Product(
        id=new_product_id,
        name="test-name",
        description="test-description",
        createDate=current_time,
        lastUpdateDate=current_time,
    )

    # Act
    with unit_of_work:
        unit_of_work.products.add(new_product)
        unit_of_work.commit()
 
    # Assert

  4. Crie uma classe de adaptador na pasta app/adapters. Use a classe abstrata da pasta ports como classe base.

  5. Execute o teste de unidade para verificar se ele falha, porque ainda não há lógica.

    python -m pytest
Desenvolvedor de aplicativos

Implemente adaptadores secundários.

  1. Implemente a lógica no arquivo do adaptador recém-criado.

  2. Atualize as afirmações do teste.

    # Assert
    with unit_of_work_readonly:
        product_from_db = unit_of_work_readonly.products.get(new_product_id)

    assertpy.assert_that(product_from_db).is_not_none()
    assertpy.assert_that(product_from_db.dict()).is_equal_to(
        {
            "id": new_product_id,
            "name": "test-name",
            "description": "test-description",
            "createDate": current_time,
            "lastUpdateDate": current_time,
        }
    )

  3. Execute o teste de unidade para verificar o sucesso.

    python -m pytest
Desenvolvedor de aplicativos

Escreva end-to-end testes.

  1. Crie um arquivo de teste na pasta app/entrypoints/api/tests usando test_ como prefixo do nome do arquivo. 

  2. Crie uma configuração de contexto do Lambda que será usada pelo teste para chamar o Lambda.

    @pytest.fixture
def lambda_context():
    @dataclass
    class LambdaContext:
        function_name: str = "test"
        memory_limit_in_mb: int = 128
        invoked_function_arn: str = "arn:aws:lambda:eu-west-1:809313241:function:test"
        aws_request_id: str = "52fdfc07-2182-154f-163f-5f0f9a621d72"

 return LambdaContext()

  3. Crie um método de teste para a invocação da API.

    def test_create_product(lambda_context):
    # Arrange
    name = "TestName"
    description = "Test description"
    request = api_model.CreateProductRequest(name=name, description=description)

    minimal_event = api_gateway_proxy_event.APIGatewayProxyEvent(
        {
            "path": "/products",
            "httpMethod": "POST",
            "requestContext": {  # correlation ID
                "requestId": "c6af9ac6-7b61-11e6-9a41-93e8deadbeef"
            },
            "body": json.dumps(request.dict()),
        }
    )

    create_product_func_mock = unittest.mock.create_autospec(
        spec=create_product_command_handler.handle_create_product_command
    )
    handler.create_product_command_handler.handle_create_product_command = (
        create_product_func_mock
    )

    # Act
    handler.handler(minimal_event, lambda_context)

  4. Execute o teste de unidade para verificar se ele falha, porque ainda não há lógica.

    python -m pytest
Desenvolvedor de aplicativos

Implemente adaptadores primários.

  1. Crie uma função para a lógica de negócios da API e declare-a como um recurso da API.

    @tracer.capture_method
@app.post("/products")
@utils.parse_event(model=api_model.CreateProductRequest, app_context=app)
def create_product(
    request: api_model.CreateProductRequest,
) -> api_model.CreateProductResponse:
    """Creates a product."""
...
    nota

    Todos os decoradores que você vê são recursos da biblioteca AWS Lambda Powertools for Python. Para obter detalhes, consulte o site Powertools do AWS Lambda para Python.

  2. Implemente a lógica da API.

    id=create_product_command_handler.handle_create_product_command(
        command=create_product_command.CreateProductCommand(
            name=request.name,
            description=request.description,
        ),
        unit_of_work=unit_of_work,
    )
    response = api_model.CreateProductResponse(id=id)
    return response.dict()

  3. Execute o teste de unidade para verificar o sucesso.

    python -m pytest
Desenvolvedor de aplicativos

Recursos relacionados

Guia do APG

Referências da AWS

Ferramentas

IDEs