ここから始める Powertools for AWS Lambda (Python) - Validation 編 -
はじめに
こんにちは。
株式会社 WHERE のプロダクト開発部で主にインフラを担当している遠藤です。
みなさん元気に Lambda Life を送っていますか??
Lambda 関数の開発に限らずですが、入力データのバリデーションは非常に重要です。
API Gateway 経由のリクエストや SQS メッセージなど、外部からのデータを安全に処理するためには、適切なバリデーション処理が欠かせません。
しかし、毎回手動でバリデーション処理を実装するのは大変で、コードの可読性や保守性にも課題を感じたことのある方も多いのではないでしょうか。
そんな課題を解決してくれるのが、Powertools for AWS Lambda の Validation 機能です。
今回はこの機能について、実際の使用イメージと共に紹介します。
なお、今後も Powertools for AWS Lambda の各種機能について継続的にブログを書いていく予定なので、興味のある方は楽しみにしていただけると嬉しいです。
※ 過去のブログ記事はコチラ👇
Powertools for AWS Lambda とは
Powertools for AWS Lambda は、サーバーレスベストプラクティスの実装と開発者の生産性向上を支援する開発者ツールキットです。
段階的に導入できる柔軟な設計により、必要な機能から順次採用することができます。
Python, TypeScript, Java, .NET で提供されており、AWS Well-Architected Serverless Lens に基づく推奨事項を簡単に実装できます。
Validation 機能とは
Validation 機能 は、JSON Schema を使用して Lambda 関数の入出力データを自動的にバリデーションするユーティリティです。
主な特徴
インバウンドイベントとアウトバウンドレスポンスのバリデーション
- Lambda 関数の入力データと出力データの両方を検証
バリデーション適用前のイベントアンラップ
- JMESPath を使用してネストされたイベント構造から必要なデータを抽出
主要イベントソース用のビルトインエンベロープ
- API Gateway, SQS, EventBridge 等の一般的なイベントソースに対応
提供されている機能
Validator decorator
-
@validatorデコレータによる関数レベルでの自動バリデーション
Validate function
-
validate()関数によるプログラマティックなバリデーション実行
Unwrapping events prior to validation
- JMESPath 式を使用したイベントの特定部分の抽出
Built-in envelopes
- API Gateway, SQS, EventBridge 等の主要イベントソース用事前定義エンベロープ
Validating custom formats
- JSON Schema の標準フォーマット以外のカスタムフォーマット定義
- AWS Account ID 等の AWS 固有フォーマットへの対応
Built-in JMESPath functions
Validating with external references
- JSON Schema の
$refキーワードによる外部スキーマ参照
JSON Schema とは
JSON Schema は、JSON データの構造と制約を定義するための標準仕様です。
主な特徴
- 型安全性: データ型の厳密な定義
- 制約定義: 文字列長, 数値範囲, 配列サイズ等の制約
- 必須フィールド: 必須項目の明確な定義
- ネストされた構造: 複雑なオブジェクト構造の定義
- 再利用性: スキーマの部品化と再利用
基本的な JSON Schema の例:
{
"type": "object",
"properties": {
"name": {
"type": "string",
"minLength": 1,
"maxLength": 100
},
"age": {
"type": "integer",
"minimum": 0,
"maximum": 150
},
"email": {
"type": "string",
"format": "email"
}
},
"required": ["name", "email"],
"additionalProperties": false
}
実践: Lambda 関数の構築・実装
ここからは、実際に Lambda 関数を構築・実装するイメージを紹介します。
※ プロジェクトの完全なコードは コチラ。
| 使用ツール | バージョン |
|---|---|
| aws-cdk-cli | 2.1021.0 |
| aws-cdk-lib | 2.199.0 |
| aws-cli | 2.13.32 |
| node | 22.13.1 |
| python | 3.11.8 |
| uv | 0.6.14 |
0. プロジェクトの初期化
# プロジェクトディレクトリの作成
mkdir validation && \
cd validation
# AWS CDK プロジェクトの初期化
cdk init app --language python
# プロジェクトのセットアップ
source .venv/bin/activate && \
uv init && \
uv add --group infra -r requirements.txt && \
uv add --dev -r requirements-dev.txt && \
mkdir lambda && \
touch lambda/function_before.py lambda/function_after.py lambda/schemas.py && \
rm requirements.txt requirements-dev.txt main.py
主なプロジェクト構成は以下のようになります:
validation/
├── validation/
│ └── validation_stack.py # AWS CDK スタック定義
├── lambda/ # Lambda 関数のソースコード
│ ├── function_after.py # Validation 機能使用版
│ ├── function_before.py # Validation 機能未使用版
│ └── schemas.py # JSON Schema 定義
├── app.py # AWS CDK アプリケーションのエントリーポイント
├── cdk.json # AWS CDK 設定ファイル
├── pyproject.toml # 依存関係管理ファイル
└── uv.lock # 依存関係管理ファイル
1. 依存関係の設定
# Powertools for AWS Lambda の追加
uv add --group lambda aws-lambda-powertools
2. JSON Schema の定義
# ユーザー情報用のスキーマ
USER_SCHEMA = {
"type": "object",
"properties": {
"name": {
"type": "string",
"minLength": 1,
"maxLength": 100
},
"email": {
"type": "string",
"format": "email"
},
"age": {
"type": "integer",
"minimum": 0,
"maximum": 150
}
},
"required": ["name", "email"],
"additionalProperties": True
}
# レスポンス用のスキーマ
RESPONSE_SCHEMA = {
"type": "object",
"properties": {
"statusCode": {
"type": "integer",
"maxLength": 3
},
"body": {
"type": "object"
}
},
"required": ["statusCode", "body"],
"additionalProperties": False
}
3. Lambda 関数の実装 (Validation 機能未使用版)
import json
from datetime import datetime, timezone
from typing import Any, Dict, List
from aws_lambda_powertools.utilities.typing import LambdaContext
def lambda_handler(event: Dict[str, Any], context: LambdaContext) -> Dict[str, Any]:
"""Validation 機能未使用"""
body: dict[str, Any] = json.loads(event["body"])
try:
# 手動でのバリデーション処理
validation_errors = validate_user_data(body)
if validation_errors:
return {
"statusCode": 400,
"body": json.dumps({"errors": validation_errors}),
}
# ビジネスロジックの実行
result = process_user_data(body)
return {
"statusCode": 200,
"body": json.dumps(result),
}
except Exception as e:
return {
"statusCode": 500,
"body": json.dumps({"error": str(e)}),
}
def validate_user_data(data: Dict[str, Any]) -> List[str]:
"""ユーザーデータの手動バリデーション"""
errors = []
# 必須フィールドの確認
if "name" not in data:
errors.append("Missing required field: name")
elif not isinstance(data["name"], str) or len(data["name"].strip()) == 0:
errors.append("Name must be a non-empty string")
elif len(data["name"]) > 100:
errors.append("Name must be less than 100 characters")
if "email" not in data:
errors.append("Missing required field: email")
elif not isinstance(data["email"], str):
errors.append("Email must be a string")
elif "@" not in data["email"]:
errors.append("Email format is invalid")
# 年齢の検証 (任意フィールド)
if "age" in data:
age = data["age"]
if not isinstance(age, int):
errors.append("Age must be an integer")
elif age < 0 or age > 150:
errors.append("Age must be between 0 and 150")
return errors
def process_user_data(user_data: Dict[str, Any]) -> Dict[str, Any]:
"""ユーザーデータの処理"""
return {
"message": "User data processed successfully",
"user_id": "user_12345",
"timestamp": datetime.now(timezone.utc).isoformat(),
"processed_data": {
"name": user_data["name"],
"email": user_data["email"],
"age": user_data.get("age"),
}
}
4. Lambda 関数の実装 (Validation 機能使用版)
from datetime import datetime, timezone
from typing import Any, Dict
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.utilities.validation import validator
from schemas import USER_SCHEMA, RESPONSE_SCHEMA
@validator(inbound_schema=USER_SCHEMA, outbound_schema=RESPONSE_SCHEMA, envelope="powertools_json(body)")
def lambda_handler(event: Dict[str, Any], context: LambdaContext) -> Dict[str, Any]:
"""Validation 機能使用"""
try:
# バリデーション済みのデータを使用してビジネスロジックを実行
result = process_user_data(event)
return {
"statusCode": 200,
"body": result,
}
except Exception as e:
return {
"statusCode": 500,
"body": {"error": str(e)},
}
def process_user_data(user_data: Dict[str, Any]) -> Dict[str, Any]:
"""ユーザーデータの処理"""
return {
"message": "User data processed successfully",
"user_id": "user_12345",
"timestamp": datetime.now(timezone.utc).isoformat(),
"processed_data": {
"name": user_data["name"],
"email": user_data["email"],
"age": user_data.get("age"),
}
}
5. AWS CDK スタックの定義
from aws_cdk import CfnOutput, Stack
from aws_cdk import aws_lambda as _lambda
from constructs import Construct
class ValidationStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# Lambda Layer: Powertools for AWS Lambda (Python)
# Layer ARN: https://docs.powertools.aws.dev/lambda/python/latest/#lambda-layer
layer = _lambda.LayerVersion.from_layer_version_arn(
self,
"PowertoolsLayer",
layer_version_arn="arn:aws:lambda:ap-northeast-1:017000801446:layer:AWSLambdaPowertoolsPythonV3-python311-x86_64:20"
)
# Lambda 関数: Validation 機能未使用版
function_before = _lambda.Function(
self,
"FunctionBefore",
runtime=_lambda.Runtime.PYTHON_3_11,
handler="function_before.lambda_handler",
code=_lambda.Code.from_asset("lambda"),
layers=[layer],
)
# Lambda 関数: Validation 機能使用版
function_after = _lambda.Function(
self,
"FunctionAfter",
runtime=_lambda.Runtime.PYTHON_3_11,
handler="function_after.lambda_handler",
code=_lambda.Code.from_asset("lambda"),
layers=[layer],
)
CfnOutput(
self,
"FunctionBeforeName",
value=function_before.function_name,
export_name="FunctionBeforeName",
)
CfnOutput(
self,
"FunctionAfterName",
value=function_after.function_name,
export_name="FunctionAfterName",
)
6. デプロイ・動作確認
# デプロイの実行
cdk deploy --require-approval never
# テスト用のペイロード作成
cat > test_valid_user.json << 'EOF'
{
"httpMethod": "POST",
"path": "/users",
"body": "{\"name\":\"John Doe\",\"email\":\"john@example.com\",\"age\":25,\"profile\":{\"bio\":\"Software engineer\",\"website\":\"https://johndoe.com\"},\"interests\":[\"programming\",\"music\"]}",
"isBase64Encoded": false
}
EOF
cat > test_invalid_user.json << 'EOF'
{
"httpMethod": "POST",
"path": "/users",
"body": "{\"name\":\"\",\"email\":\"invalid-email\",\"age\":15}",
"isBase64Encoded": false
}
EOF
# Validation 機能未使用版のテスト (有効なデータ)
aws lambda invoke \
--function-name <Before 関数名> \
--cli-binary-format raw-in-base64-out \
--payload file://test_valid_user.json \
response_before_valid.json && \
cat response_before_valid.json
# Validation 機能未使用版のテスト (無効なデータ)
aws lambda invoke \
--function-name <Before 関数名> \
--cli-binary-format raw-in-base64-out \
--payload file://test_invalid_user.json \
response_before_invalid.json && \
cat response_before_invalid.json
# Validation 機能使用版のテスト (有効なデータ)
aws lambda invoke \
--function-name <After 関数名> \
--cli-binary-format raw-in-base64-out \
--payload file://test_valid_user.json \
response_after_valid.json && \
cat response_after_valid.json
# Validation 機能使用版のテスト (無効なデータ)
aws lambda invoke \
--function-name <After 関数名> \
--cli-binary-format raw-in-base64-out \
--payload file://test_invalid_user.json \
response_after_invalid.json && \
cat response_after_invalid.json
7. リソースのクリーンアップ
cdk destroy -f
Validation 機能を使用するメリット
宣言的バリデーション
- JSON Schema による明確なデータ構造定義
- バリデーションロジックの再利用性
自動エラーハンドリング
- 詳細なバリデーションエラーメッセージ
- 一貫したエラーレスポンス形式
可読性の向上
- 複雑な手動バリデーション処理の簡略化
まとめ
本記事では、Powertools for AWS Lambda (Python) の Validation 機能について紹介しました。
JSON Schema を使用した宣言的バリデーション、Event Source への自動対応、一貫したエラーハンドリングなど、安全で保守性の高い Lambda 関数開発において非常に有用だと思います。
Powertools for AWS Lambda に興味のある方、Lambda 関数でのデータバリデーションに課題を感じている方の参考になれば嬉しいです。
今後も Powertools for AWS Lambda の各種機能についてブログを書いていく予定ですので、お楽しみに。
参考資料
- Powertools for AWS Lambda (Python) 公式ドキュメント
- Validation 機能ドキュメント
- Powertools for AWS 公式 GitHub Repository
- JSON Schema 公式サイト
最後に
株式会社 WHERE (旧: 株式会社 Penetrator) は、シリーズ A ラウンドにおいて総額 5.5 億円の資金調達を実施し、不動産テック業界における更なる成長を目指して採用活動を一層強化しています。
エンジニア, デザイナー, カスタマーサクセス, BizDev, 営業, マーケティングなど、事業拡大を支える多様なポジションで共に挑戦していただける方を待っています!!
▽ 会社のカルチャーを知りたい方はこちら
▽ 募集職種を知りたい方はこちら
Discussion