翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
AWS Lambda を使用して六角形アーキテクチャで Python プロジェクトを構築する
作成者: Furkan Oruc (AWS)、Dominik Goby (AWS)、Darius Kunce (AWS)、および Michal Ploski (AWS)
環境：PoC またはパイロット | テクノロジー: ソフトウェア開発とテスト、クラウドネイティブ、コンテナとマイクロサービス、サーバーレス、モダナイゼーション | AWS サービス： Amazon DynamoDB、AWS Lambda、Amazon API Gateway |
[概要]
このパターンは、AWS Lambda を使用して Python プロジェクトを六角形アーキテクチャで構築する方法を示しています。このパターンでは、Infrastructure as Code (IaC) ツールとして AWS Cloud Development Kit (AWS CDK)、REST API として Amazon API Gateway、パーシスタンスレイヤーとして Amazon DynamoDB を使用しています。六角形アーキテクチャは、ドメイン主導型の設計原則に従っています。六角形アーキテクチャでは、ソフトウェアはドメイン、ポート、アダプターの 3 つのコンポーネントで構成されます。六角形アーキテクチャとその利点の詳細については、「 で六角形アーキテクチャを構築する」ガイドを参照してください。
前提条件と制限
前提条件
製品バージョン
アーキテクチャ
ターゲットテクノロジースタック
ターゲットテクノロジースタックは、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 は、製品エンティティに対する 5 つの操作をサポートします。
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 API Gateway は、デベロッパーがあらゆる規模で API の公開、保守、モニタリング、セキュリティ保護を簡単に行えるフルマネージドサービスです。
Amazon DynamoDB は、あらゆる規模で高性能アプリケーションを実行できるように設計された、完全マネージド型のサーバーレスのキーバリュー型 NoSQL データベースです。
AWS Lambda は、サーバーレスのイベント駆動型のコンピューティングサービスで、サーバーのプロビジョニングや管理を行わなくても、実質あらゆるタイプのアプリケーションやバックエンドサービスのコードを実行できます。200 以上の Software as a Service (SaaS) アプリケーションから Lambda 関数を起動できます。お支払いいただくのは使用した分のみです。
ツール
このパターンでは、コード開発のバージョン管理システムとして Git が使用されます。
このパターンのプログラミング言語には Python が使用されています。Python は、高レベルのデータ構造とオブジェクト指向プログラミングへのアプローチを提供します。AWS Lambda には Python サービスの操作を簡素化する組み込みの Python ランタイムが用意されています。
このパターンの開発とテストには、Visual Studio Code が IDE として使用されています。Python 開発をサポートする任意の IDE (AWS Cloud9 や などPyCharm) を使用できます。
AWS Cloud Development Kit (AWS CDK) は、使い慣れたプログラミング言語を使用してクラウドアプリケーションリソースを定義できるオープンソースのソフトウェア開発フレームワークです。このパターンでは、CDK を使用してクラウドインフラストラクチャをコードとして記述し、デプロイします。
Poetry はパターン内の依存関係を管理するために使用されます。
Docker は AWS CDK が Lambda パッケージとレイヤーを構築するために使用します。
Code
このパターンのコードは、 GitHub Lambda の 16 進アーキテクチャのサンプルリポジトリにあります。
ベストプラクティス
実稼働環境でこのパターンを使用するには、次のベストプラクティスに従ってください。
このパターンでは、AWS X-Ray を使用して、アプリケーションのエントリポイント、ドメイン、アダプタを通じてリクエストをトレースします。AWS X-Ray は、開発者がボトルネックを特定して高いレイテンシーを洗い出し、アプリケーションのパフォーマンスを向上させるのに役立ちます。
エピック
| タスク | 説明 | 必要なスキル |
|---|
独自のリポジトリを作成します。 | にログインします GitHub。 新しいレポジトリを作成します。手順については、「」のGitHub ドキュメントを参照してください。 このパターンのサンプルリポジトリをクローン作成して、アカウントの新しいリポジトリにプッシュします。
| アプリ開発者 |
依存関係をインストールします。 | Poetry をインストールします。 pip install poetry
ルートディレクトリからパッケージをインストールします。次のコマンドで、アプリケーションと AWS CDK パッケージをインストールします。これにより、ユニットテストの実行に必要な開発パッケージもインストールされます。インストールされたパッケージはすべて新しい仮想環境に置かれます。 poetry install
インストールされたパッケージをグラフィカルに表示するには、次のコマンドを実行します。 poetry show --tree
すべての依存関係を更新します。 poetry update
新しく作成した仮想環境内で新しいシェルを開きます。ここにインストール済みの依存関係がすべて含まれています。 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 コードを使用する。 | Poetry によって管理されている仮想環境の Python インタープリターを選択してください。 テストエクスプローラーからテストを実行します。
| アプリ開発者 |
ユニットテストを実行する。オプション 2: シェルコマンドを使用する。 | 仮想環境内で新しいシェルを起動します。 poetry shell
ルートディレクトリから pytest のコマンドを実行します。 python -m pytest
または、Poetry からコマンドを直接実行できます。 poetry run python -m pytest
| アプリ開発者 |
| タスク | 説明 | 必要なスキル |
|---|
一時的な認証情報をリクエストします。 | cdk deploy の実行時にシェルに AWS 認証情報を保存するには、AWS IAM アイデンティティセンター (AWS シングルサインオンの後継) を使用して一時的な認証情報の作成を行います。手順については、ブログ記事「 IAM アイデンティティセンター で CLI を使用するための短期認証情報を取得する方法」を参照してください。
| アプリ開発者、AWS DevOps |
アプリケーションをデプロイします。 | AWS CDK v2 をインストールします。 npm install -g aws-cdk
詳細については、 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 オペレーションおよびリクエスト/レスポンスメッセージの詳細については、 GitHub リポジトリの readme ファイルの API 使用セクションを参照してください。 | アプリ開発者、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 の記事を参照してください。 API エンドポイントにリクエストを送信するには、Postman を使用します。
| アプリ開発者、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."""
...
注: 表示されるデコレータはすべて Python 用 AWS Lambda Powertools ライブラリの機能です。詳細については、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
