本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
使用 Lambda 在六角形架构中构建 Python 项目 AWS
由 Furkan Oruc (AWS)、Dominik Goby ()、Darius Kunce (AWS) 和 Michal Ploski (AWS) 创作 AWS
摘要
此模式展示了如何使用 Lambd AWS a 在六角形架构中构建 Python 项目。该模式使用AWS云开发套件 (AWSCDK) 作为基础设施即代码 (IaC) 工具，使用亚马逊API网关作为持久层，使用亚马逊 DynamoDB 作为持久层。REST API六边形架构遵循域驱动设计原则。在六边形架构中，软件由三个组件组成：域、端口和适配器。有关六边形架构及其优势的详细信息，请参阅《构建六边形架构》指南。AWS
先决条件和限制
先决条件
产品版本
架构
目标技术堆栈
目标技术堆栈由使用API网关、Lambda 和 DynamoDB 的 Python 服务组成。该服务使用 DynamoDB 适配器保存数据。它提供了使用 Lambda 作为入口点的函数。该服务使用 Amazon API Gateway 来公开 RESTAPI. API使用 Id AWS entity and Access Management (IAM) 对客户端进行身份验证。
目标架构
为说明实现方式，此模式部署了无服务器目标架构。客户端可以向API网关终端节点发送请求。 APIGateway 将请求转发到实现六角形架构模式的目标 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
AWS 服务
Amazon API Gateway 是一项完全托管的服务，可让开发人员轻松创建、发布、维护、监控和保护APIs任何规模。
Amaz on DynamoDB 是一个完全托管的无服务器键值数据库，专为运行SQL任何规模的高性能应用程序而设计。
AWSLambda 是一种无服务器、事件驱动的计算服务，它允许您为几乎任何类型的应用程序或后端服务运行代码，而无需预置或管理服务器。您可以从 200 多个服务和软件即AWS服务 (SaaS) 应用程序中启动 Lambda 函数，并且只需按实际用量付费。
工具
代码
此模式的代码可在 GitHub Lambda 六边形架构示例存储库中找到。
最佳实践
要在生产环境使用此模式，请遵循以下最佳实践：
此模式使用 AWSX- Ray 通过应用程序的入口点、域和适配器跟踪请求。 AWSX-Ray 可帮助开发人员识别瓶颈并确定高延迟，从而提高应用程序性能。
操作说明
| 任务 | 描述 | 所需技能 |
|---|
创建您自己的存储库。 | | 应用程序开发人员 |
安装依赖项。 | 安装 Poetry。 pip install poetry
从根目录安装程序包。以下命令安装应用程序和AWSCDK软件包。它还安装运行单元测试所需开发包。所有已安装的程序包都放置在新的虚拟环境中。 poetry install
要查看已安装程序包的图形表示，请运行以下命令。 poetry show --tree
更新所有依赖项。 poetry update
在新创建的虚拟环境中打开新 Shell。它包含所有已安装的依赖项。 poetry shell
| 应用程序开发人员 |
配置您的 IDE。 | 我们推荐使用 Visual Studio 代码，但你可以选择任何支持 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
| 应用程序开发人员 |
| 任务 | 描述 | 所需技能 |
|---|
请求临时凭证。 | 要在运行时在 shell 上拥有AWS证书cdk deploy，请使用 Ident AWS IAM ity Center（AWS单点登录的继任者）创建临时证书。有关说明，请参阅博客文章如何检索短期凭证以CLI用于 Ident AWS IAM ity Center。 | 应用程序开发者，AWS DevOps |
部署 应用程序。 | 安装 AWS CDK v2。 npm install -g aws-cdk
有关更多信息，请参阅AWSCDK文档。 引导AWSCDK进入你的账户和区域。 cdk bootstrap aws://12345678900/us-east-1 --profile aws-profile-name
使用AWS配置文件将应用程序部署为AWS CloudFormation 堆栈。 cdk deploy --profile aws-profile-name
| 应用程序开发者，AWS DevOps |
测试选项 1：使用控制台。API | 使用API网关控制台进行测试API。有关API操作和请求/响应消息的更多信息，请参阅存储库中自述文件的 “API用法” 部分。 GitHub | 应用程序开发者，AWS DevOps |
测试选项 2：使用 Postman。API | 如果您要使用 Postman 这样的工具，请执行以下操作： 安装 Postman 作为独立应用程序或浏览器扩展程序。 复制API网关URL的终端节点。它将采用以下格式。 https://{api-id}.execute-api.{region}.amazonaws.com/{stage}/{path}
在 “授权” 选项卡中配置AWS签名。有关说明，请参阅 reAWS: Post 中关于激活网关IAM身份验证的API文章。REST APIs 使用 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 库来模拟AWS服务。 @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."""
...
实现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参考文献
工具
IDEs
