本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
使用 AWS Lambda 在六角形架構中建構 Python 專案
由富爾坎奧魯克（AWS），多米尼克戈比（AWS），大流士昆斯（AWS）和米哈爾普洛斯基（AWS）創建
環境：PoC 或試點 | 技術：軟體開發與測試、雲端原生、容器與微服務、無伺服器、現代化 | AWS 服務：Amazon DynamoDB 支援；AWS Lambda；Amazon API Gateway |
Summary
此模式示範如何使用 AWS Lambda 在六角形架構中建構 Python 專案。該模式使用 AWS Cloud Development Kit (AWS CDK) 做為基礎設施即程式碼 (IaC) 工具，使用 Amazon API Gateway 做為其餘 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
AWS 服務
Amazon API Gateway 是一項全受管服務，可讓開發人員輕鬆建立、發佈、維護、監控和保護任何規模的 API。
Amazon DynamoDB 是全受管、無伺服器、金鑰值 NoSQL 資料庫，專為執行任何規模的高效能應用程式而設計。
AWS Lambda 是一種無伺服器、事件驅動的運算服務，可讓您針對幾乎任何類型的應用程式或後端服務執行程式碼，而無需佈建或管理伺服器。您可以從 200 多個 AWS 服務和軟體即服務 (SaaS) 應用程式啟動 Lambda 函數，而且只需按使用量付費。
工具
Code
此模式的程式碼可在 GitHub Lambda 六角形架構範例存放庫中取得。
最佳實務
若要在生產環境中使用此模式，請遵循下列最佳作法：
此病毒碼使用 AWS X-Ray 透過應用程式的入口點、網域和轉接器追蹤請求。AWS X-Ray 可協助開發人員識別瓶頸並判斷高延遲情況，以提升應用程式效能。
史诗
| 任務 | 描述 | 所需技能 |
|---|
創建您自己的存儲庫。 | | 應用程式開發人員 |
安裝依存項目。 | 安裝詩歌。 pip install poetry
從根目錄安裝套件。下列命令會安裝應用程式和 AWS CDK 套件。它也會安裝執行單元測試所需的開發套件。所有已安裝的套件都會放置在新的虛擬環境中。 poetry install
若要查看已安裝套件的圖形表示，請執行下列命令。 poetry show --tree
更新所有相依性。 poetry update
在新建立的虛擬環境中開啟新的 shell。它包含所有已安裝的依賴項。 poetry shell
| 應用程式開發人員 |
設定您的 IDE。 | 我們建議您使用視覺工作室程式碼，但您可以使用任何支援 Python 的 IDE。下面的步驟是視覺工作室代碼。 更新.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：使用視覺工作室代碼。 | 選擇由詩歌管理的虛擬環境的 Python 解釋器。 從測試資源管理器運行測試。
| 應用程式開發人員 |
運行單元測試，選項 2：使用 shell 命令。 | 在虛擬環境中啟動新的 shell。 poetry shell
從根目錄執行pytest命令。 python -m pytest
或者，您也可以直接從 Poews 執行命令。 poetry run python -m pytest
| 應用程式開發人員 |
| 任務 | 描述 | 所需技能 |
|---|
請求臨時登入資料。 | 若要在執行時在命令介面上擁有 AWS 登入資料cdk deploy，請使用 AWS IAM 身分中心 (AWS Single Sign-On 的後續任務) 建立臨時登入資料。如需指示，請參閱如何擷取短期登入資料以搭配 AWS IAM 身分中心使用 CLI 的部落格文章。 | AWS 應用程式開發人員 DevOps |
部署應用程式。 | 安裝 AWS CDK 第 2 版。 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 安裝為獨立應用程式或瀏覽器擴充功能 複製 API Gateway 的端點 URL。它將採用以下格式。 https://{api-id}.execute-api.{region}.amazonaws.com/{stage}/{path}
在授權索引標籤中設定 AWS 簽名。如需指示，請參閱 AWS RE：張貼有關為 API Gateway REST API 啟用 IAM 身份驗證的文章。 使用郵遞員將要求傳送至您的 API 端點。
| AWS 應用程式開發人員 DevOps |
| 任務 | 描述 | 所需技能 |
|---|
為業務域編寫單元測試。 | 使用檔案名稱前置詞在app/domain/tests資料夾中建立 Python test_ 檔案。 使用下列範例建立新的測試方法來測試新的商務邏輯。 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."""
...
注意：您看到的所有裝飾器都是適用於 Python 程式庫的 AWS Lambda PowerTools 的功能。如需詳細資訊，請參閱適用於 Python 的 AWS Lambda 電源工具網站。 實作 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
