Estructure un proyecto de Python en una arquitectura hexagonal con AWS Lambda - Recomendaciones de AWS
ResumenRequisitos previos y limitacionesArquitecturaHerramientasPrácticas recomendadasEpicsRecursos relacionados

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Estructure un proyecto de Python en una arquitectura hexagonal con AWS Lambda

Creado por Furkan Oruc (AWS), Dominik Goby (AWS), Darius Kunce (AWS) y Michal Ploski (AWS)

Entorno: PoC o piloto

Tecnologías: desarrollo y pruebas de software; nativo en la nube; contenedores y microservicios; sin servidor; modernización

Servicios de AWS: Amazon DynamoDB; AWS Lambda; Amazon API Gateway

Resumen

Este patrón muestra cómo estructurar un proyecto de Python en una arquitectura hexagonal mediante AWS Lambda. El patrón utiliza el AWS Cloud Development Kit (AWS CDK) como herramienta de infraestructura como código (IaC), Amazon API Gateway como REST API y Amazon DynamoDB como capa de persistencia. La arquitectura hexagonal sigue los principios de diseño basados en el dominio. En la arquitectura hexagonal, el software consta de tres componentes: dominio, puertos y adaptadores. Para obtener información detallada sobre las arquitecturas hexagonales y sus ventajas, consulte la guía Creación de arquitecturas hexagonales en AWS.

Requisitos previos y limitaciones

Requisitos previos 

Versiones de producto

  • Git versión 2.24.3 o posterior

  • Python versión 3.7 o posterior

  • AWS CDK v2

  • Poetry versión 1.1.13 o posterior

  • AWS Lambda Powertools para Python, versión 1.25.6 o posterior

  • pytest versión 7.1.1 o posterior

  • Moto versión 3.1.9 o posterior

  • versión 1.9.0 o posterior de pydantic

  • Boto3 versión 1.22.4 o posterior

  • mypy-boto3-dynamodb versión 1.24.0 o posterior

Arquitectura

Pila de tecnología de destino

La pila de tecnología de destino consiste en un servicio de Python que utiliza API Gateway, Lambda y DynamoDB. El servicio utiliza un adaptador de DynamoDB para conservar los datos. Proporciona una función que utiliza Lambda como punto de entrada. El servicio usa Amazon API Gateway para exponer una REST API. La API utiliza AWS Identity and Access Management (IAM) para la autenticación de clientes.

Arquitectura de destino 

Para ilustrar la implementación, este patrón despliega una arquitectura de destino sin servidor. Los clientes pueden enviar solicitudes a un punto de conexión final de API Gateway. API Gateway reenvía la solicitud a la función de Lambda de destino que implementa el patrón de arquitectura hexagonal. La función de Lambda realiza operaciones de creación, lectura, actualización y eliminación (CRUD) en una tabla de DynamoDB.

Importante: este patrón se probó en un entorno PoC. Debe realizar una revisión de seguridad para identificar el modelo de amenaza y crear una base de código segura antes de implementar cualquier arquitectura en un entorno de producción.

Arquitectura objetivo para estructurar un proyecto de Python en arquitectura hexagonal

La API admite cinco operaciones en una entidad de producto:

  • GET /products devuelve todos los productos.

  • POST /products crea un nuevo producto.

  • GET /products/{id} devuelve un producto específico.

  • PUT /products/{id} actualiza un producto específico.

  • DELETE /products/{id} elimina un producto específico.

Puede utilizar la siguiente estructura de carpetas para organizar el proyecto de forma que siga el patrón de arquitectura 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

Herramientas

Servicios de AWS

  • Amazon API Gateway es un servicio completamente administrado que facilita a los desarrolladores la publicación, el mantenimiento, la supervisión y la protección de las API a cualquier escala.

  • Amazon DynamoDB es una base de datos NoSQL de valor clave, sin servidor y totalmente gestionada que está diseñada para ejecutar aplicaciones de alto rendimiento a cualquier escala.

  • AWS Lambda es un servicio de computación controlado por eventos sin servidor que permite ejecutar código para prácticamente cualquier tipo de aplicación o servicio backend, sin aprovisionar ni administrar servidores. Puede lanzar funciones de Lambda desde más de 200 aplicaciones de software como servicio (SaaS) y pagar solo por lo que utilice.

Herramientas

  • Git se utiliza como sistema de control de versiones para el desarrollo de código en este patrón.

  • Python se utiliza como lenguaje de programación para este patrón. Python proporciona estructuras de datos de alto nivel y un enfoque de la programación orientada a objetos. AWS Lambda proporciona un tiempo de ejecución de Python integrado que simplifica el funcionamiento de los servicios de Python.

  • Visual Studio Code se utiliza como IDE para desarrollar y probar este patrón. Puede usar cualquier IDE que sea compatible con el desarrollo de Python (por ejemplo, AWS Cloud9 o PyCharm).

  • AWS Cloud Development Kit (AWS CDK) es un marco de desarrollo de software de código abierto que le permite definir los recursos de su aplicación en la nube utilizando lenguajes de programación conocidos. Este patrón utiliza la CDK para escribir e implementar la infraestructura de la nube como código.

  • Poetry se utiliza para gestionar las dependencias del patrón.

  • AWS CDK utiliza Docker para crear el paquete y la capa Lambda.

Código

El código de este patrón está disponible en el repositorio de ejemplos de arquitectura hexagonal GitHub Lambda.

Prácticas recomendadas

Para usar este patrón en un entorno de producción, siga estas prácticas recomendadas:

Este patrón utiliza AWS X-Ray para rastrear las solicitudes a través del punto de entrada, el dominio y los adaptadores de la aplicación. AWS X-Ray ayuda a los desarrolladores a identificar los cuellos de botella y determinar las latencias altas para mejorar el rendimiento de las aplicaciones.

Epics

TareaDescripciónHabilidades requeridas

Cree su propio repositorio.

  1. Inicie sesión en. GitHub

  2. Crear un nuevo repositorio. Para obtener instrucciones, consulte la GitHub documentación.

  3. Clone e inserte el repositorio de muestras de este patrón en el nuevo repositorio de su cuenta.

Desarrollador de aplicaciones

Instale las dependencias.

  1. Instale Poetry.

    pip install poetry

  2. Instale los paquetes desde el directorio raíz. El siguiente comando instala la aplicación y los paquetes de AWS CDK. También instala los paquetes de desarrollo necesarios para ejecutar las pruebas unitarias. Todos los paquetes instalados se colocan en un nuevo entorno virtual.

    poetry install

  3. Para ver una representación gráfica de los paquetes instalados, ejecute el siguiente comando.

    poetry show --tree

  4. Actualice todas las dependencias.

    poetry update

  5. Abra un nuevo intérprete de comandos en el entorno virtual recién creado. Contiene todas las dependencias instaladas.

    poetry shell
Desarrollador de aplicaciones

Configure su IDE.

Recomendamos Visual Studio Code, pero puede usar cualquier IDE de su elección que sea compatible con Python. Los siguientes pasos son para Visual Studio Code.

  1. Actualizar el archivo .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. Cree un archivo .env en el directorio raíz del proyecto. Esto garantiza que el directorio raíz del proyecto esté incluido en el directorio PYTHONPATH para que pytest pueda encontrarlo y descubrir correctamente todos los paquetes.

    PYTHONPATH=.
Desarrollador de aplicaciones

Ejecute pruebas unitarias, opción 1: utilice Visual Studio Code.

  1. Elija el intérprete de Python del entorno virtual gestionado por Poetry.

  2. Ejecute las pruebas desde Test Explorer.

Desarrollador de aplicaciones

Ejecute pruebas unitarias, opción 2: utilice comandos del intérprete de comandos.

  1. Inicie un nuevo intérprete de comandos en el entorno virtual.

    poetry shell

  2. Ejecute el comando pytest desde el directorio raíz.

    python -m pytest

    Otra posibilidad es ejecutar el comando directamente desde Poetry.

    poetry run python -m pytest
Desarrollador de aplicaciones
TareaDescripciónHabilidades requeridas

Solicitar credenciales temporales.

Para tener credenciales de AWS en el intérprete de comandos cuando ejecute cdk deploy, cree credenciales temporales mediante AWS IAM Identity Center (sucesor de AWS Single Sign-On). Para obtener instrucciones, consulte la entrada del blog How to retrieve short-term credentials for CLI use with AWS IAM Identity Center.

Desarrollador de aplicaciones, AWS DevOps

Implemente la aplicación .

  1. Instale la v2 de AWS CDK.

    npm install -g aws-cdk

    Para obtener más información, consulte la documentación de AWS CDK.

  2. Arranque la AWS CDK en su cuenta y región.

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

  3. Implemente la aplicación como una CloudFormation pila de AWS mediante un perfil de AWS.

    cdk deploy --profile aws-profile-name
Desarrollador de aplicaciones, AWS DevOps

Pruebe la API, opción 1: utilice la consola.

Utilice la consola de API Gateway para probar la API. Para obtener más información sobre las operaciones de la API y los mensajes de solicitud/respuesta, consulte la sección de uso de la API del archivo readme del repositorio. GitHub

Desarrollador de aplicaciones, AWS DevOps

Pruebe la API, opción 2: utilice Postman.

Si quiere utilizar una herramienta como Postman:

  1. Instale Postman como una aplicación independiente o una extensión del navegador.

  2. Copie la URL del punto de conexión de la API Gateway. Deberá tener el siguiente formato:

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

  3. Configure la firma de AWS en la pestaña de autorización. Para obtener instrucciones, consulte el artículo de AWS Re:post sobre activación de la autenticación de IAM para las REST API de API Gateway.

  4. Utilice Postman para enviar solicitudes al punto de conexión de la API.

Desarrollador de aplicaciones, AWS DevOps
TareaDescripciónHabilidades requeridas

Redacte pruebas unitarias para el dominio empresarial.

  1. Cree un archivo de Python en la carpeta app/domain/tests con el prefijo del nombre de archivo test_.

  2. Cree un nuevo método de prueba para probar la nueva lógica empresarial mediante el siguiente ejemplo.

    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. Cree una clase de comando en la carpeta app/domain/commands. 

  4. Si la funcionalidad es nueva, cree un código auxiliar para el controlador de comandos de la carpeta app/domain/command_handlers.

  5. Ejecute la prueba unitaria para comprobar si falla, ya que todavía no existe una lógica empresarial.

    python -m pytest
Desarrollador de aplicaciones

Implemente comandos y controladores de comandos.

  1. Implemente la lógica empresarial en el archivo de controlador de comandos recién creado. 

  2. Para cada dependencia que interactúe con sistemas externos, declare una clase abstracta en la carpeta 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. Actualice la firma del controlador de comandos para aceptar las dependencias recién declaradas utilizando la clase de puerto abstracta como anotación de tipo.

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

  4. Actualice la prueba unitaria para simular el comportamiento de todas las dependencias declaradas para el controlador 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. Actualice la lógica de aserción en la prueba para comprobar las invocaciones de dependencia 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. Ejecute la prueba unitaria para comprobar que funciona correctamente.

    python -m pytest
Desarrollador de aplicaciones

Escriba pruebas de integración para los adaptadores secundarios.

  1. Cree un archivo de prueba en la carpeta app/adapters/tests utilizando test_ como prefijo de nombre de archivo.

  2. Utilice la biblioteca Moto para simular los servicios de AWS.

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

  3. Cree un nuevo método de prueba para una prueba de integración del 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. Cree una clase de adaptador en la carpeta app/adapters. Utilice la clase abstracta de la carpeta de puertos como clase base.

  5. Ejecute la prueba unitaria para ver si falla, porque todavía no hay lógica.

    python -m pytest
Desarrollador de aplicaciones

Implemente adaptadores secundarios.

  1. Implemente la lógica en el archivo de adaptador recién creado.

  2. Actualice las afirmaciones de las pruebas.

    # 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. Ejecute la prueba unitaria para comprobar que funciona correctamente.

    python -m pytest
Desarrollador de aplicaciones

end-to-end Redacte pruebas.

  1. Cree un archivo de prueba en la carpeta app/entrypoints/api/tests utilizando test_ como prefijo de nombre de archivo. 

  2. Cree un elemento de contexto de Lambda que la prueba utilizará para llamar a 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. Cree un método de prueba para la invocación de la 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. Ejecute la prueba unitaria para ver si falla, porque todavía no hay lógica.

    python -m pytest
Desarrollador de aplicaciones

Implemente los adaptadores principales.

  1. Cree una función para la lógica empresarial de la API y declare como un recurso de la 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 los decoradores que ve son características de la biblioteca AWS Lambda Powertools para Python. Para obtener más información, consulte el sitio web de AWS Lambda Powertools para Python.

  2. Implemente la lógica de la 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. Ejecute la prueba unitaria para comprobar que funciona correctamente.

    python -m pytest
Desarrollador de aplicaciones

Recursos relacionados

Guía de APG

Referencias de AWS

Herramientas

IDEs