Structure a Python project in hexagonal architecture using AWS Lambda

Log in to GitHub.

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

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

Install Poetry.

pip install poetry

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

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

poetry show --tree

Update all dependencies.

poetry update

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

poetry shell

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

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

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

Run tests from Test Explorer.

Start a new shell within the virtual environment.

poetry shell

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

Install the AWS CDK v2.

npm install -g aws-cdk

For more information, see the AWS CDK documentation.

Bootstrap the AWS CDK into your account and Region.

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

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

cdk deploy --profile aws-profile-name

Install Postman as a standalone application or browser extension.

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}

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.

Use Postman to send requests to your API endpoint.

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

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

Create a command class in the app/domain/commands folder. 

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

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

python -m pytest

Implement business logic in the newly created command handler file. 

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

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

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
    )

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

Run the unit test to see it succeed.

python -m pytest

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

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

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

Create an adapter class in the app/adapters folder. Use the abstract class from the ports folder as a base class.

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

python -m pytest

Implement logic in the newly created adapter file.

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

Run the unit test to see it succeed.

python -m pytest

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

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

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)

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

python -m pytest

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.

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

Run the unit test to see it succeed.

python -m pytest