Structure a Python project in hexagonal architecture using AWS Lambda - AWS Prescriptive Guidance
SummaryPrerequisites and limitationsArchitectureToolsBest practicesEpicsRelated resources

Structure a Python project in hexagonal architecture using AWS Lambda

Created by Furkan Oruc (AWS), Dominik Goby (AWS), Darius Kunce (AWS), and Michal Ploski (AWS)

Environment: PoC or pilot

Technologies: Software development & testing; Cloud-native; Containers & microservices; Serverless; Modernization

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

Summary

This pattern shows how to structure a Python project in hexagonal architecture by using AWS Lambda. The pattern uses the AWS Cloud Development Kit (AWS CDK) as the infrastructure as code (IaC) tool, Amazon API Gateway as the REST API, and Amazon DynamoDB as the persistence layer. Hexagonal architecture follows domain-driven design principles. In hexagonal architecture, software consists of three components: domain, ports, and adapters. For detailed information about hexagonal architectures and their benefits, see the guide Building hexagonal architectures on AWS.

Prerequisites and limitations

Prerequisites 

Product versions

  • Git version 2.24.3 or later

  • Python version 3.7 or later

  • AWS CDK v2

  • Poetry version 1.1.13 or later

  • AWS Lambda Powertools for Python version 1.25.6 or later

  • pytest version 7.1.1 or later

  • Moto version 3.1.9 or later

  • pydantic version 1.9.0 or later

  • Boto3 version 1.22.4 or later

  • mypy-boto3-dynamodb version 1.24.0 or later

Architecture

Target technology stack  

The target technology stack consists of a Python service that uses API Gateway, Lambda, and DynamoDB. The service uses a DynamoDB adapter to persist data. It provides a function that uses Lambda as the entry point. The service uses Amazon API Gateway to expose a REST API. The API uses AWS Identity and Access Management (IAM) for the authentication of clients.

Target architecture 

To illustrate the implementation, this pattern deploys a serverless target architecture. Clients can send requests to an API Gateway endpoint. API Gateway forwards the request to the target Lambda function that implements the hexagonal architecture pattern. The Lambda function performs create, read, update, and delete (CRUD) operations on a DynamoDB table.

Important: This pattern was tested in a PoC environment. You must conduct a security review to identify the threat model and create a secure code base before you deploy any architecture to a production environment.

Target architecture for structuring a Python project in hexagonal architecture

The API supports five operations on a product entity:

  • GET /products returns all products.

  • POST /products creates a new product.

  • GET /products/{id} returns a specific product.

  • PUT /products/{id} updates a specific product.

  • DELETE /products/{id} deletes a specific product.

You can use the following folder structure to organize your project to follow the hexagonal architecture pattern:  

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

Tools

AWS services

  • Amazon API Gateway is a fully managed service that makes it easy for developers to create, publish, maintain, monitor, and secure APIs at any scale.

  • Amazon DynamoDB is a fully managed, serverless, key-value NoSQL database that is designed to run high-performance applications at any scale.

  • AWS Lambda is a serverless, event-driven compute service that lets you run code for virtually any type of application or backend service without provisioning or managing servers. You can launch Lambda functions from over 200 AWS services and software as a service (SaaS) applications, and only pay for what you use.

Tools

  • Git  is used as the version control system for code development in this pattern.

  • Python is used as the programming language for this pattern. Python provides high-level data structures and an approach to object-oriented programming. AWS Lambda provides a built-in Python runtime that simplifies the operation of Python services.

  • Visual Studio Code is used as the IDE for development and testing for this pattern. You can use any IDE that supports Python development (for example, AWS Cloud9 or PyCharm).

  • AWS Cloud Development Kit (AWS CDK) is an open-source software development framework that lets you define your cloud application resources by using familiar programming languages. This pattern uses the CDK to write and deploy cloud infrastructure as code.

  • Poetry is used to manage dependencies in the pattern.

  • Docker is used by the AWS CDK to build the Lambda package and layer.

Code 

The code for this pattern is available in the GitHub Lambda hexagonal architecture sample repository.

Best practices

To use this pattern in a production environment, follow these best practices:

This pattern uses AWS X-Ray to trace requests through the application’s entry point, domain, and adapters. AWS X-Ray helps developers identify bottlenecks and determine high latencies to improve application performance.

Epics

TaskDescriptionSkills required

Create your own repository.

  1. Log in to GitHub.

  2. Create a new repository. For instructions, see the GitHub documentation.

  3. Clone and push the sample repository for this pattern into the new repository in your account.

App developer

Install dependencies.

  1. Install Poetry.

    pip install poetry

  2. Install packages from the root directory. The following command installs the application and AWS CDK packages. It also installs development packages that are required for running unit tests. All installed packages are placed in a new virtual environment.

    poetry install

  3. To see a graphical representation of the installed packages, run the following command.

    poetry show --tree

  4. Update all dependencies.

    poetry update

  5. Open a new shell within the newly created virtual environment. It contains all installed dependencies.

    poetry shell
App developer

Configure your IDE.

We recommend Visual Studio Code, but you can use any IDE of your choice that supports Python. The following steps are for Visual Studio Code.

  1. Update the .vscode/settings file.

    {
    "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. Create an .env file in the root directory of the project. This ensures that the root directory of the project is included in the PYTHONPATH so that pytest can find it and properly discover all packages.

    PYTHONPATH=.
App developer

Run unit tests, option 1: Use Visual Studio Code.

  1. Choose the Python interpreter of the virtual environment that’s managed by Poetry.

  2. Run tests from Test Explorer.

App developer

Run unit tests, option 2: Use shell commands.

  1. Start a new shell within the virtual environment.

    poetry shell

  2. Run the pytest command from the root directory.

    python -m pytest

    Alternatively you can run the command directly from Poetry.

    poetry run python -m pytest
App developer
TaskDescriptionSkills required

Request temporary credentials.

To have AWS credentials on the shell when you run cdk deploy, create temporary credentials by using AWS IAM Identity Center (successor to AWS Single Sign-On). For instructions, see the blog post How to retrieve short-term credentials for CLI use with AWS IAM Identity Center.

App developer, AWS DevOps

Deploy the application.

  1. Install the AWS CDK v2.

    npm install -g aws-cdk

    For more information, see the AWS CDK documentation.

  2. Bootstrap the AWS CDK into your account and Region.

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

  3. Deploy the application as an AWS CloudFormation stack by using an AWS profile.

    cdk deploy --profile aws-profile-name
App developer, AWS DevOps

Test the API, option 1: Use the console.

Use the API Gateway console to test the API. For more information about API operations and request/response messages, see the API usage section of the readme file in the GitHub repository.

App developer, AWS DevOps

Test the API, option 2: Use Postman.

If you want to use a tool such as Postman:

  1. Install Postman as a standalone application or browser extension.

  2. Copy the endpoint URL for the API Gateway. It will be in the following format.

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

  3. Configure the AWS signature in the authorization tab. For instructions, see the AWS re:Post article on activating IAM authentication for API Gateway REST APIs.

  4. Use Postman to send requests to your API endpoint.

App developer, AWS DevOps
TaskDescriptionSkills required

Write unit tests for the business domain.

  1. Create a Python file in the app/domain/tests folder by using the test_ file name prefix.

  2. Create a new test method to test the new business logic by using the following example.

    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. Create a command class in the app/domain/commands folder. 

  4. If the functionality is new, create a stub for the command handler in the app/domain/command_handlers folder.

  5. Run the unit test to see it fail, because there is still no business logic.

    python -m pytest
App developer

Implement commands and command handlers.

  1. Implement business logic in the newly created command handler file. 

  2. For every dependency that interacts with external systems, declare an abstract class in the app/domain/ports folder.

    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. Update the command handler signature to accept the newly declared dependencies by using the abstract port class as type annotation.

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

  4. Update the unit test to simulate the behavior of all declared dependencies for the command handler.

        # 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. Update the assertion logic in the test to check for the expected dependency invocations.

      # 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. Run the unit test to see it succeed.

    python -m pytest
App developer

Write integration tests for secondary adapters.

  1. Create a test file in the app/adapters/tests folder by using test_ as a file name prefix.

  2. Use the Moto library to mock AWS services.

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

  3. Create a new test method for an integration test of the adapter.

    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. Create an adapter class in the app/adapters folder. Use the abstract class from the ports folder as a base class.

  5. Run the unit test to see it fail, because there is still no logic.

    python -m pytest
App developer

Implement secondary adapters.

  1. Implement logic in the newly created adapter file.

  2. Update test assertions.

    # 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. Run the unit test to see it succeed.

    python -m pytest
App developer

Write end-to-end tests.

  1. Create a test file in the app/entrypoints/api/tests folder by using test_ as a file name prefix. 

  2. Create a Lambda context fixture that will be used by the test to call 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. Create a test method for the API invocation.

    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. Run the unit test to see it fail, because there is still no logic.

    python -m pytest
App developer

Implement primary adapters.

  1. Create a function for API business logic and declare it as an API resource.

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

    Note: All decorators you see are features of the AWS Lambda Powertools for Python library. For details, see the AWS Lambda Powertools for Python website.

  2. Implement the API logic.

    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. Run the unit test to see it succeed.

    python -m pytest
App developer

Related resources

APG guide

AWS References

Tools

IDEs