本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
使用 AWS Lambda 以六边形架构构建 Python 项目
创建者：Furkan Oruc (AWS)、Dominik Goby (AWS)、Darius Kunce (AWS) 和 Michal Ploski (AWS)
环境：PoC 或试点 | 技术：软件开发和测试；云原生；容器和微服务；无服务器；现代化 | Amazon Web Services：Amazon DynamoDB；AWS Lambda；Amazon API Gateway |
总结
此模式展示了如何使用 AWS Lambda 以六边形架构构建 Python 项目。该模式使用 AWS Cloud Development Kit (AWS CDK) 作为基础设施即代码（IaC）工具，使用 Amazon API Gateway 作为 REST API，使用 Amazon DynamoDB 作为持久层。六边形架构遵循域驱动设计原则。在六边形架构中，软件由三个组件组成：域、端口和适配器。有关六边形架构及其优势的详细信息，请参阅指南在 AWS 上构建六边形架构。
先决条件和限制
先决条件
产品版本
架构
目标技术堆栈
目标技术堆栈由使用 API Gateway、Lambda 和 DynamoDB 的 Python 服务构成。该服务使用 DynamoDB 适配器保存数据。它提供了使用 Lambda 作为入口点的函数。该服务使用 Amazon API Gateway 公开 REST API。API 使用 AWS Identity and Access Management (IAM) 对客户端执行身份验证。
目标架构
为说明实现方式，此模式部署了无服务器目标架构。客户端可向 API Gateway 端点发送请求。API Gateway 将请求转发至实现六边形架构模式的目标 Lambda 函数。Lambda 函数可对 DynamoDB 表执行创建、读取、更新和删除（CRUD）操作。
重要提示：此模式已在 PoC 环境中进行了测试。在将任何架构部署至生产环境之前，您必须进行安全审查以识别威胁模型并创建安全的代码库。 |
|---|
该 API 支持对产品实体的五种操作：
GET /products 返回所有产品。
POST /products 创建新产品。
GET /products/{id} 返回特定产品。
PUT /products/{id} 更新特定产品。
DELETE /products/{id} 删除特定产品。
您可使用以下文件夹结构来组织项目，以遵循六边形架构模式：
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
Amazon Web Services
Amazon API Gateway 是一项完全托管的服务，使开发人员可以轻松创建、发布、维护、监控和保护任何规模的 API。
Amazon DynamoDB 是一个完全托管的无服务器键值的 NoSQL 数据库，专为运行任何规模的高性能应用程序而设计。
AWS Lambda 是一项无服务器、事件驱动计算服务，让您能够为几乎任何类型的应用程序或后端服务运行代码，而无需预调配或管理服务器。您可以从 200 多种 Amazon Web Services 和软件即服务（SaaS）应用程序启动 Lambda 函数，并且只需为您使用的部分付费。
工具
代码
此模式的代码可在 GitHub Lambda 六边形架构示例存储库中找到。
最佳实践
要在生产环境使用此模式，请遵循以下最佳实践：
此模式使用 AWS X-Ray 通过应用程序的入口点、域和适配器跟踪请求。AWS X-Ray 帮助开发人员识别瓶颈并确定高延迟以提高应用程序性能。
操作说明
| 任务 | 描述 | 所需技能 |
|---|
创建您自己的存储库。 | | 应用程序开发人员 |
安装依赖项。 | 安装 Poetry。 pip install poetry
从根目录安装程序包。以下命令安装应用程序和 AWS CDK 包。它还安装运行单元测试所需的开发包。所有已安装的程序包都放置在新的虚拟环境中。 poetry install
要查看已安装程序包的图形表示，请运行以下命令。 poetry show --tree
更新所有依赖项。 poetry update
在新创建的虚拟环境中打开新 Shell。它包含所有已安装的依赖项。 poetry shell
| 应用程序开发人员 |
配置您的 IDE。 | 我们推荐 Visual Studio Code，但您可以使用您选择的任何支持 Python 的 IDE。以下步骤适用于 Visual Studio Code。 更新 .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",
}
在项目的根目录中创建 .env文件。这样可以确保项目的根目录包含在 PYTHONPATH中，以便 pytest可以找到它并正确发现所有程序包。 PYTHONPATH=.
| 应用程序开发人员 |
运行单元测试，选项 1：使用 Visual Studio Code。 | 选择由 Poetry 管理虚拟环境的 Python 解释器。 从测试资源管理器中运行测试。
| 应用程序开发人员 |
运行单元测试，选项 2：使用 Shell 命令。 | 在虚拟环境中启动一个新 Shell。 poetry shell
从根目录运行 pytest命令。 python -m pytest
或者，您可以直接从 Poetry 中运行该命令。 poetry run python -m pytest
| 应用程序开发人员 |
| 任务 | 描述 | 所需技能 |
|---|
请求临时凭证。 | 要在运行 cdk deploy时在 Shell 中使用 AWS 凭证，请使用 AWS IAM Identity Center（AWS Single Sign-On 的后继任务）创建临时凭证。有关说明，请参阅博客文章如何检索短期凭证，以便在 AWS IAM Identity Center 使用 CLI。 | AWS 应用程序开发人员 DevOps |
部署 应用程序。 | 安装 AWS CDK v2。 npm install -g aws-cdk
有关更多信息，请参阅 AWS CDK 文档。 将 AWS CDK 引导到您的账户和区域。 cdk bootstrap aws://12345678900/us-east-1 --profile aws-profile-name
使用 AWS 配置文件将应用程序部署为 AWS CloudFormation 堆栈。 cdk deploy --profile aws-profile-name
| AWS 应用程序开发人员 DevOps |
测试 API，选项 1：使用控制台。 | 使用 API Gateway 控制台测试 API。有关 API 操作和请求/响应消息的更多信息，请参阅存储库中自述文件的 API 用法部分。 GitHub | AWS 应用程序开发人员 DevOps |
测试 API，选项 2：使用 Postman。 | 如果您要使用 Postman 这样的工具，请执行以下操作： 安装 Postman 作为独立应用程序或浏览器扩展程序。 复制 API Gateway 的端点 URL。它将采用以下格式。 https://{api-id}.execute-api.{region}.amazonaws.com/{stage}/{path}
在授权选项卡中配置 AWS 签名。有关说明，请参阅关于激活 API Gateway REST API 的 IAM 身份验证的 AWS re:Post 文章。 使用 Postman 向 API 端点发送请求。
| AWS 应用程序开发人员 DevOps |
| 任务 | 描述 | 所需技能 |
|---|
为业务域编写单元测试。 | 通过使用 test_文件名前缀在 app/domain/tests文件夹中创建 Python 文件。 通过使用以下示例创建新的测试方法，以测试新的业务逻辑。 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
在 app/domain/commands文件夹中创建命令类。 如果该功能是新增的，请在 app/domain/command_handlers文件夹中为命令处理程序创建一个存根。 运行单元测试以查看其失败问题，因为仍然没有业务逻辑。 python -m pytest
| 应用程序开发人员 |
实施命令和命令处理程序。 | 在新创建的命令处理程序文件中实施业务逻辑。 对于每个与外部系统交互的依赖项，请在 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:
...
使用抽象端口类作为类型注释，更新命令处理程序签名以接受新声明的依赖项。 def handle_create_product_command(
command: create_product_command.CreateProductCommand,
unit_of_work: unit_of_work.UnitOfWork,
) -> str:
...
更新单元测试以模拟命令处理程序的所有声明的依赖项的行为。 # 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
)
更新测试中的断言逻辑以检查是否有预期的依赖项调用。 # 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")
运行单元测试，以查看它是否成功。 python -m pytest
| 应用程序开发人员 |
为辅助适配器编写集成测试。 | 通过使用 test_作为文件名前缀在 app/adapters/tests文件夹中创建测试文件。 使用 Moto 库模拟 Amazon Web Services。 @pytest.fixture
def mock_dynamodb():
with moto.mock_dynamodb():
yield boto3.resource("dynamodb", region_name="eu-central-1")
为适配器的集成测试创建新测试方法。 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
在 app/adapters文件夹中创建适配器类。使用 ports 文件夹中的抽象类作为基础类。 运行单元测试，看看它是否失败，因为仍然没有逻辑。 python -m pytest
| 应用程序开发人员 |
实施辅助适配器。 | 在新创建的适配器文件中实施逻辑。 更新测试断言。 # 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,
}
)
运行单元测试，以查看它是否成功。 python -m pytest
| 应用程序开发人员 |
编写 end-to-end 测试。 | 通过使用 test_作为文件名前缀在 app/entrypoints/api/tests文件夹中创建测试文件。 创建 Lambda 上下文固定装置，测试将使用该装置来调用 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()
为 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)
运行单元测试，看看它是否失败，因为仍然没有逻辑。 python -m pytest
| 应用程序开发人员 |
实施主适配器。 | 为 API 业务逻辑创建函数，并将其声明为 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."""
...
注意：您看到的所有装饰器都是 AWS Lambda Powertools for Python 库的功能。有关详细信息，请访问 AWS Lambda Powertools for Python 网站。 实施 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()
运行单元测试，以查看它是否成功。 python -m pytest
| 应用程序开发人员 |
相关资源
APG 指南
AWS 参考
工具
IDE
