Estruture um projeto Python em arquitetura hexagonal usando Lambda AWS

Faça login em GitHub.

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

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

Instale o Poetry.

pip install poetry

Instale pacotes do diretório raiz. O comando a seguir instala o aplicativo e AWS CDK os pacotes. 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

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

poetry show --tree

Atualizar todas as dependências.

poetry update

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

poetry shell

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

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=. 

Selecione o interpretador Python do ambiente virtual gerenciado pelo Poetry.

Execute testes no Test Explorer.

Inicie um novo shell no ambiente virtual.

poetry shell

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

Instale a AWS CDK v2.

npm install -g aws-cdk

Para obter mais informações, consulte a AWSCDKdocumentação.

Inicialize o AWS CDK em sua conta e região.

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

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

cdk deploy --profile aws-profile-name

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

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

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

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

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

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

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

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

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

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

python -m pytest

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

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:
        ...

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:
    ...

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
    )

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

Execute o teste de unidade para verificar o sucesso.

python -m pytest

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

Use a biblioteca Moto para simular AWS serviços.

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

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

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

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

python -m pytest

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

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,
        }
    )

Execute o teste de unidade para verificar o sucesso.

python -m pytest

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

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()

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

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)

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

python -m pytest

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

@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."""
...

Observação: todos os decoradores que você vê são recursos da biblioteca AWS Lambda Powertools for Python. Para obter detalhes, consulte o site AWSLambda Powertools for Python.

Implemente a API lógica.

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()

Execute o teste de unidade para verificar o sucesso.

python -m pytest