Estructura un proyecto de Python en una arquitectura hexagonal usando Lambda AWS - 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.

Estructura un proyecto de Python en una arquitectura hexagonal usando Lambda AWS

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

Entorno: PoC o piloto

Tecnologías: aplicaciones web y móviles; contenedores y microservicios; tecnología sin servidor; modernización

AWSservicios: Amazon DynamoDBAWS; Lambda; Amazon Gateway API

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 (AWSCDK) como herramienta de infraestructura como código (IaC), Amazon API Gateway como capa de persistencia 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 Construir arquitecturas hexagonales sobre ellas. 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

  • AWSCDKv2

  • Poetry versión 1.1.13 o posterior

  • AWSLambda 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 tecnológica 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 utiliza Amazon API Gateway para exponer un RESTAPI. APIUtiliza AWS Identity and Access Management (IAM) para la autenticación de los 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 final de API Gateway. APIGateway reenvía la solicitud a la función Lambda de destino que implementa el patrón de arquitectura hexagonal. La función 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

APIAdmite 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

AWSservicios

  • Amazon API Gateway es un servicio totalmente gestionado que facilita a los desarrolladores la creación, publicación, mantenimiento, supervisión y protección APIs a cualquier escala.

  • Amazon DynamoDB es una base de datos sin valor clave totalmente gestionada, sin servidores y diseñada para ejecutar SQL aplicaciones de alto rendimiento a cualquier escala.

  • AWSLambda es un servicio informático sin servidor y basado en eventos que le permite ejecutar código para prácticamente cualquier tipo de aplicación o servicio de backend sin aprovisionar ni administrar servidores. Puede lanzar funciones Lambda desde más de 200 AWS servicios y 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. AWSLambda proporciona un motor de ejecución de Python integrado que simplifica el funcionamiento de los servicios de Python.

  • Se utiliza Visual Studio Code IDE para el desarrollo y las pruebas de este patrón. Puede usar cualquiera IDE que soporte el desarrollo de Python (por ejemplo, PyCharm).

  • AWSCloud Development Kit (AWSCDK) es un marco de desarrollo de software de código abierto que le permite definir los recursos de sus aplicaciones en la nube mediante lenguajes de programación conocidos. Este patrón lo utiliza CDK para escribir e implementar la infraestructura de la nube como código.

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

  • El utiliza Docker AWS CDK 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 AWSX-Ray para rastrear las solicitudes a través del punto de entrada, el dominio y los adaptadores de la aplicación. AWSX-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 AWS CDK paquetes. 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 cualquiera IDE de sus preferencias que sea compatible con Python. Los siguientes pasos son para Visual Studio Code.

  1. Actualice 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

Solicite credenciales temporales.

Para tener AWS credenciales en el shell al ejecutarcdk deploy, cree credenciales temporales mediante AWS IAM Identity Center (sucesor de AWS Single Sign-On). Para obtener instrucciones, consulte la entrada del blog Cómo recuperar credenciales de corta duración para CLI usarlas con AWS IAM Identity Center.

Desarrollador de aplicaciones, AWS DevOps

Implemente la aplicación .

  1. Instala la AWS CDK v2.

    npm install -g aws-cdk

    Para obtener más información, consulte la AWSCDKdocumentación.

  2. Incorpórelo AWS CDK a su cuenta y región.

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

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

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

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

Utilice la consola API Gateway para probar elAPI. Para obtener más información sobre API las operaciones y los mensajes de solicitud/respuesta, consulte la sección de API uso 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 el punto final de URL la API puerta de enlace. Deberá tener el siguiente formato:

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

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

  4. Utilice Postman para enviar solicitudes a su punto final. 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. Usa la biblioteca Moto para simular AWS servicios.

    @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

Escribe end-to-end 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. Crea un método de prueba para la API invocación.

    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 API empresarial y declárela como un 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."""
...

    Nota: Todos los decoradores que ve son funciones de la biblioteca AWS Lambda Powertools para Python. Para obtener más información, consulte el AWSsitio web Lambda Powertools para Python.

  2. Implemente la lógicaAPI.

    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

APGguía

AWSReferencias

Herramientas

IDEs