Structurer un projet Python dans une architecture hexagonale à l'aide d'AWS Lambda - Recommandations AWS
RécapitulatifConditions préalables et limitationsArchitectureOutilsBonnes pratiquesÉpopéesRessources connexes

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Structurer un projet Python dans une architecture hexagonale à l'aide d'AWS Lambda

Créée par Furkan Oruc (AWS), Dominik Goby (AWS), Darius Kunce (AWS) et Michal Ploski (AWS)

Environnement : PoC ou pilote

Technologies : DevelopmentAndTesting ; CloudNative ; Conteneurs et microservices ; Sans serveur ; Modernisation

Services AWS : Amazon DynamoDB ; AWS Lambda ; Amazon API Gateway

Récapitulatif

Ce modèle montre comment structurer un projet Python dans une architecture hexagonale à l'aide d'AWS Lambda. Le modèle utilise l'AWS Cloud Development Kit (AWS CDK) comme outil d'infrastructure en tant que code (iAc), Amazon API Gateway comme API REST et Amazon DynamoDB comme couche de persistance. L'architecture hexagonale suit les principes de conception axés sur le domaine. Dans une architecture hexagonale, le logiciel comprend trois composants : le domaine, les ports et les adaptateurs. Pour obtenir des informations détaillées sur les architectures hexagonales et leurs avantages, consultez le guide Création d'architectures hexagonales sur AWS.

Conditions préalables et limitations

Prérequis

Versions du produit

  • Git version 2.24.3 ou ultérieure

  • Python version 3.7 ou ultérieure

  • Kit de développement logiciel AWS version 2

  • Poetry version 1.1.13 ou ultérieure

  • AWS Lambda Powertools pour Python version 1.25.6 ou ultérieure

  • pytest version 7.1.1 ou ultérieure

  • Moto version 3.1.9 ou ultérieure

  • version 1.9.0 ou ultérieure de pydantic

  • Boto3 version 1.22.4 ou ultérieure

  • mypy-boto3-dynamodb version 1.24.0 ou ultérieure

Architecture

Pile technologique cible

La pile technologique cible consiste en un service Python qui utilise API Gateway, Lambda et DynamoDB. Le service utilise un adaptateur DynamoDB pour conserver les données. Il fournit une fonction qui utilise Lambda comme point d'entrée. Le service utilise Amazon API Gateway pour exposer une API REST. L'API utilise AWS Identity and Access Management (IAM) pour l'authentification des clients.

Architecture cible

Pour illustrer l'implémentation, ce modèle déploie une architecture cible sans serveur. Les clients peuvent envoyer des demandes à un point de terminaison API Gateway. API Gateway transmet la demande à la fonction Lambda cible qui implémente le modèle d'architecture hexagonal. La fonction Lambda effectue des opérations de création, de lecture, de mise à jour et de suppression (CRUD) sur une table DynamoDB.

Important : Ce modèle a été testé dans un environnement PoC. Vous devez effectuer un examen de sécurité pour identifier le modèle de menace et créer une base de code sécurisée avant de déployer une architecture dans un environnement de production.

Architecture cible pour structurer un projet Python en architecture hexagonale

L'API prend en charge cinq opérations sur une entité de produit :

  • GET /productsrenvoie tous les produits.

  • POST /productscrée un nouveau produit.

  • GET /products/{id}renvoie un produit spécifique.

  • PUT /products/{id}met à jour un produit spécifique.

  • DELETE /products/{id}supprime un produit spécifique.

Vous pouvez utiliser la structure de dossiers suivante pour organiser votre projet selon le modèle d'architecture 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

Outils

Services AWS

  • Amazon API Gateway est un service entièrement géré qui permet aux développeurs de créer, publier, gérer, surveiller et sécuriser facilement des API à n'importe quelle échelle.

  • Amazon DynamoDB est une base de données NoSQL à valeur clé entièrement gérée, sans serveur, conçue pour exécuter des applications hautes performances à n'importe quelle échelle.

  • AWS Lambda est un service de calcul sans serveur piloté par les événements qui vous permet d'exécuter du code pour pratiquement n'importe quel type d'application ou de service principal sans provisionner ni gérer de serveurs. Vous pouvez lancer des fonctions Lambda à partir de plus de 200 services AWS et applications logicielles en tant que service (SaaS), et ne payer que pour ce que vous utilisez.

Outils

  • Git est utilisé comme système de contrôle de version pour le développement de code dans ce modèle.

  • Python est utilisé comme langage de programmation pour ce modèle. Python fournit des structures de données de haut niveau et une approche de la programmation orientée objet. AWS Lambda fournit un environnement d'exécution Python intégré qui simplifie le fonctionnement des services Python.

  • Visual Studio Code est utilisé comme IDE pour le développement et les tests de ce modèle. Vous pouvez utiliser n'importe quel IDE prenant en charge le développement en Python (par exemple, AWS Cloud9 ou PyCharm).

  • AWS Cloud Development Kit (AWS CDK) est un framework de développement logiciel open source qui vous permet de définir les ressources de vos applications cloud à l'aide de langages de programmation courants. Ce modèle utilise le CDK pour écrire et déployer l'infrastructure cloud sous forme de code.

  • La poésie est utilisée pour gérer les dépendances dans le modèle.

  • Docker est utilisé par le AWS CDK pour créer le package et la couche Lambda.

Code

Le code de ce modèle est disponible dans le référentiel d'exemples d'architecture hexagonale GitHub Lambda.

Bonnes pratiques

Pour utiliser ce modèle dans un environnement de production, suivez les meilleures pratiques suivantes :

Ce modèle utilise AWS X-Ray pour suivre les demandes via le point d'entrée, le domaine et les adaptateurs de l'application. AWS X-Ray aide les développeurs à identifier les goulets d'étranglement et à déterminer les latences élevées afin d'améliorer les performances des applications.

Épopées

TâcheDescriptionCompétences requises

Créez votre propre référentiel.

  1. Connectez-vous à GitHub.

  2. Créez un nouveau référentiel. Pour obtenir des instructions, consultez la GitHub documentation.

  3. Clonez et transférez le référentiel d'échantillons correspondant à ce modèle dans le nouveau référentiel de votre compte.

Développeur d'applications

Installez les dépendances.

  1. Installez Poetry.

    pip install poetry

  2. Installez les packages depuis le répertoire racine. La commande suivante installe l'application et les packages AWS CDK. Il installe également les packages de développement nécessaires à l'exécution des tests unitaires. Tous les packages installés sont placés dans un nouvel environnement virtuel.

    poetry install

  3. Pour afficher une représentation graphique des packages installés, exécutez la commande suivante.

    poetry show --tree

  4. Mettez à jour toutes les dépendances.

    poetry update

  5. Ouvrez un nouveau shell dans l'environnement virtuel nouvellement créé. Il contient toutes les dépendances installées.

    poetry shell
Développeur d'applications

Configurez votre IDE.

Nous recommandons Visual Studio Code, mais vous pouvez utiliser n'importe quel IDE de votre choix qui supporte Python. Les étapes suivantes concernent Visual Studio Code.

  1. Mettez à jour le .vscode/settings fichier.

    {
    "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. Créez un .env fichier dans le répertoire racine du projet. Cela garantit que le répertoire racine du projet est inclus dans le PYTHONPATH afin de pytest pouvoir le trouver et découvrir correctement tous les packages.

    PYTHONPATH=.
Développeur d'applications

Exécuter des tests unitaires, option 1 : utiliser Visual Studio Code.

  1. Choisissez l'interpréteur Python de l'environnement virtuel géré par Poetry.

  2. Exécutez des tests à partir de l'explorateur de tests.

Développeur d'applications

Exécuter des tests unitaires, option 2 : utiliser des commandes shell.

  1. Démarrez un nouveau shell dans l'environnement virtuel.

    poetry shell

  2. Exécutez la pytest commande depuis le répertoire racine.

    python -m pytest

    Vous pouvez également exécuter la commande directement depuis Poetry.

    poetry run python -m pytest
Développeur d'applications
TâcheDescriptionCompétences requises

Demandez des informations d'identification temporaires.

Pour avoir des informations d'identification AWS sur le shell lorsque vous exécutezcdk deploy, créez des informations d'identification temporaires à l'aide d'AWS IAM Identity Center (successeur d'AWS Single Sign-On). Pour obtenir des instructions, consultez le billet de blog Comment récupérer des informations d'identification à court terme pour une utilisation en CLI avec AWS IAM Identity Center.

Développeur d'applications, AWS DevOps

Déployez l'application.

  1. Installez le kit AWS CDK v2.

    npm install -g aws-cdk

    Pour plus d'informations, consultez la documentation AWS CDK.

  2. Intégrez le CDK AWS à votre compte et à votre région.

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

  3. Déployez l'application en tant que CloudFormation stack AWS à l'aide d'un profil AWS.

    cdk deploy --profile aws-profile-name
Développeur d'applications, AWS DevOps

Testez l'API, option 1 : utilisez la console.

Utilisez la console API Gateway pour tester l'API. Pour plus d'informations sur les opérations d'API et les messages de demande/réponse, consultez la section sur l'utilisation de l'API du fichier readme du référentiel. GitHub

Développeur d'applications, AWS DevOps

Testez l'API, option 2 : utilisez Postman.

Si vous souhaitez utiliser un outil tel que Postman :

  1. Installez Postman en tant qu'application autonome ou extension de navigateur.

  2. Copiez l'URL du point de terminaison pour l'API Gateway. Il sera présenté dans le format suivant.

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

  3. Configurez la signature AWS dans l'onglet d'autorisation. Pour obtenir des instructions, consultez l'article AWS re:Post sur l'activation de l'authentification IAM pour les API REST API Gateway.

  4. Utilisez Postman pour envoyer des demandes à votre point de terminaison d'API.

Développeur d'applications, AWS DevOps
TâcheDescriptionCompétences requises

Rédigez des tests unitaires pour le domaine commercial.

  1. Créez un fichier Python dans le app/domain/tests dossier en utilisant le préfixe du nom de test_ fichier.

  2. Créez une nouvelle méthode de test pour tester la nouvelle logique métier à l'aide de l'exemple suivant.

    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. Créez une classe de commande dans le app/domain/commands dossier. 

  4. Si la fonctionnalité est nouvelle, créez un stub pour le gestionnaire de commandes dans le app/domain/command_handlers dossier.

  5. Exécutez le test unitaire pour voir s'il échoue, car il n'existe toujours aucune logique métier.

    python -m pytest
Développeur d'applications

Implémentez des commandes et des gestionnaires de commandes.

  1. Implémentez la logique métier dans le fichier de gestionnaire de commandes nouvellement créé. 

  2. Pour chaque dépendance interagissant avec des systèmes externes, déclarez une classe abstraite dans le app/domain/ports dossier.

    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. Mettez à jour la signature du gestionnaire de commandes pour accepter les dépendances nouvellement déclarées en utilisant la classe de port abstraite comme annotation de type.

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

  4. Mettez à jour le test unitaire pour simuler le comportement de toutes les dépendances déclarées pour le gestionnaire de commandes.

        # 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. Mettez à jour la logique d'assertion dans le test pour vérifier les invocations de dépendance attendues.

      # 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. Exécutez le test unitaire pour voir s'il réussit.

    python -m pytest
Développeur d'applications

Rédigez des tests d'intégration pour les adaptateurs secondaires.

  1. Créez un fichier de test dans le app/adapters/tests dossier en l'utilisant test_ comme préfixe de nom de fichier.

  2. Utilisez la bibliothèque Moto pour simuler les services AWS.

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

  3. Créez une nouvelle méthode de test pour un test d'intégration de l'adaptateur.

    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. Créez une classe d'adaptateur dans le app/adapters dossier. Utilisez la classe abstraite du dossier ports comme classe de base.

  5. Exécutez le test unitaire pour le voir échouer, car il n'y a toujours aucune logique.

    python -m pytest
Développeur d'applications

Implémentez des adaptateurs secondaires.

  1. Implémentez la logique dans le fichier d'adaptateur nouvellement créé.

  2. Mettez à jour les assertions de test.

    # 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. Exécutez le test unitaire pour voir s'il réussit.

    python -m pytest
Développeur d'applications

Rédigez end-to-end des tests.

  1. Créez un fichier de test dans le app/entrypoints/api/tests dossier en l'utilisant test_ comme préfixe de nom de fichier. 

  2. Créez un dispositif de contexte Lambda qui sera utilisé par le test pour appeler 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. Créez une méthode de test pour l'invocation de l'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. Exécutez le test unitaire pour le voir échouer, car il n'y a toujours aucune logique.

    python -m pytest
Développeur d'applications

Implémentez les adaptateurs principaux.

  1. Créez une fonction pour la logique métier de l'API et déclarez-la en tant que ressource d'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."""
...

    Remarque : Tous les décorateurs que vous voyez sont des fonctionnalités de la bibliothèque AWS Lambda Powertools for Python. Pour plus de détails, consultez le site Web AWS Lambda Powertools for Python.

  2. Implémentez la logique de l'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. Exécutez le test unitaire pour voir s'il réussit.

    python -m pytest
Développeur d'applications

Ressources connexes

Guide APG

Références AWS

Outils

IDE