- Python
Lambda Powertools Pythonの使い方|Logger・Tracer
Powertools for AWS Lambda(Python)は、Lambda 関数に「構造化ログ」「カスタムメトリクス」「トレース」などを数行で足せる、AWS 公式のオープンソースのライブラリです。 ハンドラにデコレーターを3つ重ねると、リクエストID付きの JSON ログ、CloudWatch のメトリクス、X-Ray のトレースがまとめて出ます。
from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.logging import correlation_paths
logger = Logger() # POWERTOOLS_SERVICE_NAME を service として出す
tracer = Tracer() # X-Ray にサブセグメントを送る
metrics = Metrics() # POWERTOOLS_METRICS_NAMESPACE に EMF で出す
@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_REST)
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler(event, context):
logger.info("hello")
return {"statusCode": 200}
| 機能 | 何をするか | 追加の依存 |
|---|---|---|
| Logger | JSON の構造化ログ。関数名・リクエストID・コールドスタート・相関IDを自動で付ける | なし |
| Metrics | CloudWatch のカスタムメトリクスを EMF(ログに書くだけで指標になる形式)で出す | なし |
| Tracer | X-Ray のサブセグメント・注釈・メタデータを付ける | aws-xray-sdk([tracer]) |
| Event Handler | API Gateway・ALB・関数 URL などのルーティング | なし(入力検証は Pydantic) |
| Parser | Pydantic でイベントを型付きのモデルに変換・検証 | pydantic([parser]) |
| Idempotency | 同じリクエストの二重実行を防ぐ(DynamoDB などに結果を保存) | なし(boto3) |
| Batch | SQS・Kinesis・DynamoDB Streams の「一部だけ失敗」を返す | なし |
確認環境:aws-lambda-powertools 3.35.0・pydantic 2.14.1・aws-xray-sdk 2.15.0・boto3 1.43.112・Python 3.11(Linux)。ハンドラは Lambda の context と同じ属性を持つオブジェクトを渡して直接呼び、Logger・Metrics の出力、REST API(v1)と HTTP API(v2)のイベントでのルーティングと入力検証、Parser、Batch、moto 5.2.4 の DynamoDB を使った Idempotency を動かしました。Tracer は Lambda の環境変数を与え、UDP 2000 番で待つ擬似の X-Ray デーモンを立てて、送られる中身を受け取りました。インストール方法・環境変数・上限は Powertools の公式ドキュメント(docs.aws.amazon.com/powertools/python)とリポジトリのドキュメントで確認しています(確認日 2026年10月13日)。
本記事の内容は、ご自由にお使いください。
ご利用の際は、出典として本ページへのリンクを記載いただけますようお願いします。
(記載例)出典:株式会社RJC「Lambda Powertools Pythonの使い方|Logger・Tracer」
入れ方:pip とレイヤー
pip(デプロイパッケージに入れる)
pip install "aws-lambda-powertools" # Logger・Metrics・Event Handler・Batch・Idempotency
pip install "aws-lambda-powertools[tracer]" # Tracer を使うなら(aws-xray-sdk が入る)
pip install "aws-lambda-powertools[parser]" # Parser を使うなら(pydantic が入る)
pip install "aws-lambda-powertools[all]" # まとめて
pip install "aws-lambda-powertools[aws-sdk]" # 手元の開発用(boto3 を入れる)
ドキュメントの extras の表です。
| extras | 入るもの | 使う機能 |
|---|---|---|
[tracer] |
aws-xray-sdk |
Tracer |
[parser] |
pydantic |
Parser(Event Handler の入力検証も) |
[validation] |
fastjsonschema |
Validation(JSON Schema) |
[datamasking] |
aws-encryption-sdk・jsonpath-ng |
Data Masking |
[redis]・[valkey] |
redis・valkey-glide |
Idempotency の保存先を Redis/Valkey に |
[all] |
主な extras 全部 | — |
[aws-sdk] |
boto3 | 手元の開発(Lambda には boto3 が入っている) |
Tracer を使うのに [tracer] を入れ忘れると、Tracer() を作った時点で止まります。 3.35.0 で aws-xray-sdk を外して動かした結果です。
Traceback (most recent call last):
File "app.py", line 2, in <module>
tracer = Tracer(service="orders")
File ".../aws_lambda_powertools/tracing/tracer.py", line 193, in __init__
self.__build_config(
...
from aws_xray_sdk.core import xray_recorder # type: ignore
ModuleNotFoundError: No module named 'aws_xray_sdk'
Lambda の外(手元・テスト)でも同じです。モジュールの読み込み時に失敗するため、関数は Runtime.ImportModuleError になります。関連記事「Lambda Unable to import moduleの直し方|Python」で扱っています。
Lambda レイヤー(AWS が公開している)
ドキュメントの表(バージョン 3 系)の ARN の形です。{python_version} は python310〜python314 です。
x86_64 : arn:aws:lambda:{region}:017000801446:layer:AWSLambdaPowertoolsPythonV3-{python_version}-x86_64:37
arm64 : arn:aws:lambda:{region}:017000801446:layer:AWSLambdaPowertoolsPythonV3-{python_version}-arm64:37
末尾の数字はレイヤーのバージョンで、リリースごとに増えます。SSM パラメータ /aws/service/powertools/python/{arch}/{python版}/latest に最新の ARN が公開されているので、CloudFormation では次のように参照できます。
Layers:
- !Sub "{{resolve:ssm:/aws/service/powertools/python/x86_64/python3.13/latest}}"
レイヤーには Tracer の aws-xray-sdk などの依存もまとめて入っています(ドキュメントでは x86_64・ARM 用にコンパイル済みの依存を含む、と説明)。
SAM テンプレートの例
Globals:
Function:
Runtime: python3.13
Tracing: Active # X-Ray を有効に(Tracer に必要)
Environment:
Variables:
POWERTOOLS_SERVICE_NAME: orders
POWERTOOLS_METRICS_NAMESPACE: MyShop
POWERTOOLS_LOG_LEVEL: INFO
Resources:
OrdersFunction:
Type: AWS::Serverless::Function
Properties:
Handler: app.lambda_handler
CodeUri: src/
Layers:
- !Sub "{{resolve:ssm:/aws/service/powertools/python/x86_64/python3.13/latest}}"
Events:
Api:
Type: Api
Properties: { Path: /{proxy+}, Method: ANY }
SAM の書き方は、関連記事「AWS SAM入門|template.yamlの書き方とbuild・deploy」で扱っています。
環境変数の一覧
| 環境変数(ドキュメント) | 対象 | 既定値 |
|---|---|---|
POWERTOOLS_SERVICE_NAME |
全部(service キー・ディメンション) |
service_undefined |
POWERTOOLS_LOG_LEVEL |
Logger | INFO |
POWERTOOLS_LOGGER_LOG_EVENT |
Logger(受け取ったイベントをログに出す) | false |
POWERTOOLS_LOGGER_SAMPLE_RATE |
Logger(DEBUG にする呼び出しの割合 0〜1) | 0 |
POWERTOOLS_METRICS_NAMESPACE |
Metrics | None |
POWERTOOLS_METRICS_DISABLED |
Metrics を出さない | false |
POWERTOOLS_TRACE_DISABLED |
Tracer を止める | false |
POWERTOOLS_TRACER_CAPTURE_RESPONSE |
戻り値をメタデータに入れる | true |
POWERTOOLS_TRACER_CAPTURE_ERROR |
例外をメタデータに入れる | true |
POWERTOOLS_DEV |
開発用(ログを整形して見やすく、など) | false |
完成コード:API Gateway(REST)+ Logger・Tracer・Metrics
このあとの出力は、すべて次のコードを動かしたものです。
# app.py
from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.event_handler.exceptions import NotFoundError
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.metrics import MetricUnit
logger = Logger()
tracer = Tracer()
metrics = Metrics()
app = APIGatewayRestResolver()
ORDERS = {"1": {"id": "1", "item": "coffee", "qty": 2}}
@app.get("/orders/<order_id>")
@tracer.capture_method
def get_order(order_id: str):
order = ORDERS.get(order_id)
if order is None:
raise NotFoundError(f"order {order_id} not found")
logger.info("order found", extra={"order_id": order_id})
metrics.add_metric(name="OrderRead", unit=MetricUnit.Count, value=1)
return order
@app.post("/orders")
def create_order():
body = app.current_event.json_body
logger.append_keys(customer_id=body.get("customer_id"))
logger.info("order created")
metrics.add_dimension(name="channel", value="web")
metrics.add_metric(name="OrderCreated", unit=MetricUnit.Count, value=1)
return {"id": "2", **body}, 201 # (本文, ステータス) のタプルも返せる
@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_REST)
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler(event, context):
return app.resolve(event, context)
POWERTOOLS_SERVICE_NAME=orders・POWERTOOLS_METRICS_NAMESPACE=MyShop で、REST API(ペイロード v1)のイベントを4回渡しました。
| リクエスト | 戻り値の statusCode |
body |
|---|---|---|
GET /orders/1 |
200 | {"id":"1","item":"coffee","qty":2} |
GET /orders/9(無い ID) |
404 | {"statusCode":404,"message":"order 9 not found"} |
POST /orders |
201 | {"id":"2","customer_id":"c-001","item":"tea"} |
DELETE /orders/1(ルートが無い) |
404 | {"statusCode":404,"message":"Not found"} |
登録していないメソッドは 405 ではなく 404 Not found になりました。辞書を返すと JSON に変換され、Content-Type: application/json が付きます。
Logger:構造化 JSON と相関ID
出力の実物
GET /orders/1 のときに出た1行です(読みやすく改行しています)。
{
"level": "INFO",
"location": "get_order:21",
"message": "order found",
"timestamp": "2026-10-13 14:42:29,666+0900",
"service": "orders",
"cold_start": true,
"function_name": "orders-api",
"function_memory_size": 256,
"function_arn": "arn:aws:lambda:ap-northeast-1:123456789012:function:orders-api",
"function_request_id": "c6af9ac6-7b61-11e6-9a41-93e812345678",
"correlation_id": "req-abc-123",
"order_id": "1"
}
| キー | どこから来るか |
|---|---|
level・location・message・timestamp |
いつも出る。location は「関数名:行番号」 |
service |
POWERTOOLS_SERVICE_NAME か Logger(service=...) |
cold_start・function_* |
@logger.inject_lambda_context が context から付ける |
correlation_id |
correlation_id_path で指定したイベントの場所(ここでは API Gateway の requestContext.requestId) |
order_id |
extra= で、その1行だけに足したキー |
timestamp のオフセットは実行した環境のタイムゾーンで、手元(日本時間)では +0900 でした。Lambda の実行環境は UTC が既定のため、Lambda 上では +0000 になります。
相関ID:イベントのどこを使うか
correlation_paths に、よく使う場所が定数で用意されています(3.35.0 のソースで確認)。
| 定数 | 中身(JMESPath) |
|---|---|
API_GATEWAY_REST・API_GATEWAY_HTTP |
requestContext.requestId |
LAMBDA_FUNCTION_URL |
requestContext.requestId |
APPLICATION_LOAD_BALANCER |
headers."x-amzn-trace-id" |
APPSYNC_RESOLVER |
request.headers."x-amzn-trace-id" |
EVENT_BRIDGE |
id |
VPC_LATTICE |
headers."x-amzn-trace-id" |
フロントエンドから X-Request-Id などの独自ヘッダーを渡しているなら、correlation_id_path='headers."x-request-id"' のように JMESPath を書くか、ハンドラの中で logger.set_correlation_id(...) を呼びます。複数サービスでこの ID をそろえると、CloudWatch Logs Insights で1つのリクエストを横断して追えます。関連記事「構造化ログ(JSON)設計|必須フィールド一覧とPython・Node実装例」で、ログに入れる項目の設計を扱っています。
append_keys は次の呼び出しに残る
logger.append_keys() で足したキーは、Logger の状態として残り、次の呼び出しのログにも出ます。 Lambda は実行環境を使い回すためです。2回呼んで確かめました(1回目だけ customer_id を足す)。
# @logger.inject_lambda_context(既定)
{'message': 'hello', 'cold_start': True, 'customer_id': 'c-1'}
{'message': 'hello', 'cold_start': False, 'customer_id': 'c-1'} ← 2回目にも残った
# @logger.inject_lambda_context(clear_state=True)
{'message': 'hello2', 'cold_start': False, 'customer_id': 'c-1'}
{'message': 'hello2', 'cold_start': False} ← 消えた
ユーザーIDのように呼び出しごとに変わる値は、clear_state=True を付けるか、毎回必ず上書きします。1行だけなら extra= を使います。
例外のログ
try:
1 / 0
except ZeroDivisionError:
logger.exception("calc failed")
{"level":"ERROR","location":"<module>:15","message":"calc failed", "...":"...",
"exception":"Traceback (most recent call last):\n File \"/var/task/lg.py\", line 14, in <module>\n try: 1/0\n ~^~\nZeroDivisionError: division by zero",
"exception_name":"ZeroDivisionError",
"stack_trace":{"type":"ZeroDivisionError","value":"division by zero","module":"builtins",
"frames":[{"file":"/var/task/lg.py","line":14,"function":"<module>","statement":"try: 1/0"}]}}
トレースバックが1行の JSON に入るため、CloudWatch Logs で1件のログとして扱えます。exception_name で集計もできます。
よく使う Logger の操作
| やりたいこと | 書き方 |
|---|---|
| 1行だけキーを足す | logger.info("msg", extra={"k": v}) |
| 以降のログ全部に足す | logger.append_keys(k=v) |
| 足したキーを消す | logger.remove_keys(["k"]) |
| 呼び出しごとに状態を消す | @logger.inject_lambda_context(clear_state=True) |
| 受け取ったイベントを出す | inject_lambda_context(log_event=True) か POWERTOOLS_LOGGER_LOG_EVENT=true |
| 一部の呼び出しだけ DEBUG | POWERTOOLS_LOGGER_SAMPLE_RATE=0.1(10%) |
| 別モジュールで同じ設定を使う | Logger(child=True) |
log_event=True は、イベントに個人情報やトークンが入っていると、そのままログに残ります。本番では既定の false のまま使い、必要な項目だけを extra で出す形が安全です。
Metrics:EMF でカスタムメトリクス
出力の実物
@metrics.log_metrics は、ハンドラが終わるときに、ためたメトリクスを EMF(Embedded Metric Format)の JSON 1行として標準出力に書きます。CloudWatch Logs がこの形式を見つけると、PutMetricData を呼ばずにメトリクスとして取り込みます。
{"_aws":{"Timestamp":1791870149671,"CloudWatchMetrics":[{"Namespace":"MyShop","Dimensions":[["function_name","service"]],"Metrics":[{"Name":"ColdStart","Unit":"Count"}]}]},"function_name":"orders-api","service":"orders","ColdStart":[1.0]}
{"_aws":{"Timestamp":1791870149672,"CloudWatchMetrics":[{"Namespace":"MyShop","Dimensions":[["service"]],"Metrics":[{"Name":"OrderRead","Unit":"Count"}]}]},"service":"orders","OrderRead":[1.0]}
{"_aws":{"Timestamp":1791870149673,"CloudWatchMetrics":[{"Namespace":"MyShop","Dimensions":[["channel","service"]],"Metrics":[{"Name":"OrderCreated","Unit":"Count"}]}]},"channel":"web","service":"orders","OrderCreated":[1.0]}
| 読み方 | 内容 |
|---|---|
Namespace |
POWERTOOLS_METRICS_NAMESPACE(ここでは MyShop) |
Dimensions |
service は自動。add_dimension で足したもの(channel)が加わる |
ColdStart |
capture_cold_start_metric=True のとき、コールドスタートの1回目だけ別の行で出る。ディメンションは function_name・service |
"OrderRead":[1.0] |
値。同じ名前を何度も add_metric すると配列に並ぶ |
動かして分かった注意
| 状況 | 結果(3.35.0) |
|---|---|
| 名前空間を設定していない | ハンドラの終わりに SchemaValidationError: Must contain a metric namespace. で関数が失敗した |
| メトリクスを1つも足さずに終わった | UserWarning: No application metrics to publish. の警告だけ(失敗しない) |
| 毎回必ず出したい | @metrics.log_metrics(raise_on_empty_metrics=True) で、空なら例外にできる |
上の 404 のリクエスト(GET /orders/9)ではメトリクスを足していないため、警告が出ました。
ドキュメントにある EMF の決まりです。
| 項目(ドキュメント) | 値 |
|---|---|
| 1つの EMF に入るメトリクス | 100 個。100 個目を足した時点で書き出し、101 個目からは次の EMF に入る |
| ユーザー定義のディメンション | 29 個まで |
| ハンドラの外(グローバル)で足したメトリクス・ディメンション | コールドスタートのときだけ付く |
| 呼び出しをまたいで残すディメンション | metrics.set_default_dimensions(...) |
ディメンションに注文IDやユーザーIDのような値を入れないようにします。ディメンションの組み合わせごとに別のメトリクスになり、カスタムメトリクスの料金が増えます。ID は Logger の extra に入れ、Logs Insights で集計します。
Tracer:X-Ray のサブセグメント・注釈
コード
from aws_lambda_powertools import Tracer
tracer = Tracer(service="orders")
@tracer.capture_method
def load(order_id):
tracer.put_annotation(key="order_id", value=order_id) # 検索できる
tracer.put_metadata(key="order", value={"id": order_id, "qty": 2}) # 検索できない
return {"id": order_id}
@tracer.capture_lambda_handler
def handler(event, context):
return load("1")
X-Ray デーモンに届いた中身
Lambda と同じ環境変数(AWS_LAMBDA_FUNCTION_NAME、トレースID の _X_AMZN_TRACE_ID)を与え、UDP 2000 番で受け取った実物です。
{"format":"json","version":1}
{"id": "97d8fe857126a5fd", "name": "## __main__.load", "parent_id": "962c47dd91e1d736",
"annotations": {"order_id": "1"},
"metadata": {"orders": {"order": {"id": "1", "qty": 2}, "__main__.load response": {"id": "1"}}},
"trace_id": "1-6720a1b2-5f0c3e2d1a9b8c7d6e5f4a3b", "type": "subsegment", ...}
{"format":"json","version":1}
{"id": "962c47dd91e1d736", "name": "## handler", "parent_id": "53995c3f42cd8ad8",
"annotations": {"ColdStart": true, "Service": "orders"},
"metadata": {"orders": {"handler response": {"id": "1"}}},
"trace_id": "1-6720a1b2-5f0c3e2d1a9b8c7d6e5f4a3b", "type": "subsegment", ...}
| 分かったこと | 内容 |
|---|---|
| サブセグメント名 | ハンドラは ## handler、メソッドは ## モジュール名.関数名 |
| 自動で付く注釈 | ColdStart(真偽値)と Service |
| 戻り値 | 関数名 response としてメタデータに入る(POWERTOOLS_TRACER_CAPTURE_RESPONSE) |
| メタデータの名前空間 | サービス名(ここでは orders) |
| 親 | ハンドラの parent_id は _X_AMZN_TRACE_ID の Parent(Lambda が作るセグメント) |
大きな戻り値や個人情報を返す関数は、@tracer.capture_method(capture_response=False) にします。 X-Ray のセグメント文書は 64 KB までで、メタデータもその中に入ります。
Lambda の外では自動で止まる
| 実行場所 | Tracer の状態(3.35.0) |
|---|---|
手元・pytest(AWS_LAMBDA_FUNCTION_NAME が無い) |
無効(_is_tracer_disabled() が True)。何も送らない |
| Lambda(環境変数あり) | 有効。デーモンにサブセグメントを送る |
POWERTOOLS_TRACE_DISABLED=true |
どこでも無効 |
関数側で アクティブトレース(Tracing: Active)を有効にしないと、Lambda がトレースを作らず、X-Ray には何も出ません。 X-Ray の有効化・サービスマップ・サンプリングは、関連記事「AWS X-Rayの使い方|Lambda・APIのトレースとサービスマップ」で扱っています。boto3 の呼び出しは、Tracer を作った時点で自動でパッチされ、サブセグメントになります(Tracer(patch_modules=[...]) で絞れます)。
Event Handler:REST API と HTTP API のルーティング
どのリゾルバーを使うか
| リゾルバー(ドキュメントの表) | 対象 |
|---|---|
APIGatewayRestResolver |
API Gateway REST API(ペイロード v1) |
APIGatewayHttpResolver |
API Gateway HTTP API(ペイロード v2 が既定) |
ALBResolver |
Application Load Balancer |
LambdaFunctionUrlResolver |
Lambda 関数 URL |
VPCLatticeResolver・VPCLatticeV2Resolver |
VPC Lattice |
HttpResolverLocal(別名 HttpResolver) |
ASGI 互換。手元の HTTP サーバーで同じルートを動かす用 |
HTTP API でもペイロード v1 を選んでいるなら、APIGatewayRestResolver を使います(ドキュメントの注意)。REST API と HTTP API の違いは、関連記事「API Gateway Lambda|REST APIとHTTP APIの違い」で扱っています。
HTTP API + Pydantic での入力検証
enable_validation=True にすると、引数の型ヒント(Pydantic のモデルや Query)で本文・クエリ文字列を検証し、合わなければ 422 を返します。
from typing import Optional
from typing_extensions import Annotated
from pydantic import BaseModel, Field
from aws_lambda_powertools import Logger
from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
from aws_lambda_powertools.event_handler.openapi.params import Query
from aws_lambda_powertools.logging import correlation_paths
logger = Logger(service="orders")
app = APIGatewayHttpResolver(enable_validation=True)
class OrderIn(BaseModel):
item: str = Field(min_length=1)
qty: int = Field(gt=0, le=100)
note: Optional[str] = None
class OrderOut(OrderIn):
id: str
@app.post("/orders")
def create(order: OrderIn) -> OrderOut:
return OrderOut(id="o-1", **order.model_dump())
@app.get("/orders")
def list_orders(limit: Annotated[int, Query(gt=0, le=50)] = 10):
return {"limit": limit, "items": []}
@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_HTTP)
def lambda_handler(event, context):
return app.resolve(event, context)
HTTP API(v2)のイベントを渡した結果です。
POST /orders {"item":"tea","qty":3} → 200 {"item":"tea","qty":3,"note":null,"id":"o-1"}
POST /orders {"item":"","qty":0} → 422 {"statusCode":422,"detail":[{"loc":["body","item"],"type":"string_too_short"},{"loc":["body","qty"],"type":"greater_than"}]}
GET /orders?limit=5 → 200 {"limit":5,"items":[]}
GET /orders?limit=500 → 422 {"statusCode":422,"detail":[{"loc":["query","limit"],"type":"less_than_equal"}]}
文字列の "5" は int に変換されて通りました。エラーの detail には、どの項目が(loc)、なぜ(type)失敗したかが入ります。Pydantic の検証の書き方は、関連記事「Pydantic v2の使い方|BaseModelとバリデーションをエラー例で解説」で扱っています。
ルーティングの書き方早見表
| やりたいこと | 書き方 |
|---|---|
| パスパラメータ | @app.get("/orders/<order_id>") → 引数 order_id |
| 複数メソッド | @app.route("/orders", method=["PUT", "PATCH"]) |
| クエリ文字列 | app.current_event.get_query_string_value("limit", default_value="10") |
| ヘッダー | app.current_event.headers.get("x-api-key") |
| JSON 本文 | app.current_event.json_body |
| ステータスやヘッダーを決める | return Response(status_code=201, content_type="application/json", body=..., headers={...}) |
| エラーを返す | raise NotFoundError(...)・BadRequestError・UnauthorizedError など |
| 例外ごとに処理 | @app.exception_handler(ValueError) |
| 無いルート | @app.not_found |
| CORS | APIGatewayRestResolver(cors=CORSConfig(allow_origin="https://example.com")) |
| ファイルを分ける | Router() を作り、app.include_router(router, prefix="/orders") |
| Swagger UI | app.enable_swagger(path="/swagger")(enable_validation=True と組み合わせ) |
Parser:イベントを Pydantic で受ける
Event Handler を使わないトリガー(EventBridge・SQS・S3 など)では、Parser でイベントをモデルに変換します。エンベロープは「イベントのどこに本体があるか」を表し、EventBridgeEnvelope なら detail だけを取り出してモデルに入れます。
from pydantic import BaseModel
from aws_lambda_powertools.utilities.parser import event_parser, ValidationError
from aws_lambda_powertools.utilities.parser.envelopes import EventBridgeEnvelope
class Order(BaseModel):
order_id: str
qty: int
@event_parser(model=Order, envelope=EventBridgeEnvelope)
def lambda_handler(order: Order, context):
return {"got": order.order_id, "qty": order.qty}
detail = {"order_id":"o-1","qty":"2"} → {'got': 'o-1', 'qty': 2}
detail = {"order_id":"o-1","qty":"two"} → ValidationError
1 validation error for Order
qty
Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='two', input_type=str]
| 主なエンベロープ | 取り出す場所 |
|---|---|
EventBridgeEnvelope |
detail |
SqsEnvelope |
各レコードの body(JSON)。モデルのリストになる |
SnsEnvelope・SnsSqsEnvelope |
SNS のメッセージ、SQS 経由の SNS メッセージ |
ApiGatewayEnvelope・ApiGatewayV2Envelope |
API Gateway の body |
DynamoDBStreamEnvelope |
NewImage・OldImage |
KinesisDataStreamEnvelope |
Base64 を戻したデータ |
イベント全体の型が欲しいときは、SqsModel・EventBridgeModel などの組み込みモデルを継承し、detail などの型だけ差し替えます。
Idempotency:二重実行を防ぐ
同じ支払いIDのリクエストが2回届いても、処理は1回だけにし、2回目は保存した結果を返します。 保存先は DynamoDB(Redis・Valkey も可)です。
from aws_lambda_powertools.utilities.idempotency import (
DynamoDBPersistenceLayer, IdempotencyConfig, idempotent)
layer = DynamoDBPersistenceLayer(table_name="IdempotencyTable") # パーティションキー id(文字列)
config = IdempotencyConfig(
event_key_jmespath="powertools_json(body).payment_id", # 何を「同じリクエスト」とみなすか
expires_after_seconds=3600, # 結果を覚えておく秒数
)
@idempotent(persistence_store=layer, config=config)
def lambda_handler(event, context):
charge(...) # 1回だけ実行したい処理
return {"statusCode": 200, "body": json.dumps({"charged": True})}
moto の DynamoDB に向けて、同じ payment_id のイベントを2回渡した結果です。
1回目: {'statusCode': 200, 'body': '{"charged": true, "n": 1}'}
2回目: {'body': '{"charged": true, "n": 1}', 'statusCode': 200} ← 保存した結果
handler ran 1 times
テーブルの中身:
{'id': 'test-func.__main__.pay#1632a722e2bedc78b660140bc0cc32a6',
'expiration': '1791873843', 'status': 'COMPLETED',
'in_progress_expiration': '1791870246394',
'data': '{"body": "{\"charged\": true, \"n\": 1}", "statusCode": 200}'}
キーは「関数名.モジュール.関数#キーのハッシュ」です。手元では AWS_LAMBDA_FUNCTION_NAME が無いため test-func になりました。expiration に DynamoDB の TTL を設定すると、期限切れのレコードが消えます。処理中に同じリクエストが来ると IdempotencyAlreadyInProgressError になります。冪等性の考え方は、関連記事「冪等性とは?API設計での意味と冪等キーの実装方法」で扱っています。
| 関数の IAM 権限(DynamoDB の場合) |
|---|
dynamodb:GetItem・PutItem・UpdateItem・DeleteItem |
Batch:SQS の部分失敗
SQS から10件まとめて受け取り、1件だけ失敗したとき、失敗した1件だけをキューに戻すのが Batch です。Lambda のイベントソースマッピングで ReportBatchItemFailures を有効にしておきます。
from aws_lambda_powertools.utilities.batch import BatchProcessor, EventType, process_partial_response
from aws_lambda_powertools.utilities.data_classes.sqs_event import SQSRecord
processor = BatchProcessor(event_type=EventType.SQS)
def record_handler(record: SQSRecord):
body = json.loads(record.body)
if body["qty"] <= 0:
raise ValueError(f"bad qty {body['qty']}")
def lambda_handler(event, context):
return process_partial_response(event=event, record_handler=record_handler,
processor=processor, context=context)
m-1 {"qty":1}, m-2 {"qty":0}, m-3 {"qty":5} → {'batchItemFailures': [{'itemIdentifier': 'm-2'}]}
m-4 {"qty":0} だけ → BatchProcessingError: All records failed processing. 1 individual errors logged separately below.
全件失敗すると、戻り値ではなく例外になり、呼び出し全体が失敗します(バッチ全体が再試行の対象)。何度も失敗するメッセージは、デッドレターキューに送られるよう設定します。関連記事「SQSデッドレターキューの設定|maxReceiveCountと再処理」で扱っています。
EventType |
対象 |
|---|---|
EventType.SQS |
SQS(FIFO は SqsFifoPartialProcessor) |
EventType.KinesisDataStreams |
Kinesis Data Streams |
EventType.DynamoDBStreams |
DynamoDB Streams |
テストの書き方(pytest)
inject_lambda_context は context の属性を読むため、テストでは同じ属性を持つオブジェクトを渡します。
# test_app.py
import json
from dataclasses import dataclass
import pytest
@dataclass
class LambdaContext:
function_name: str = "orders-api"
memory_limit_in_mb: int = 256
invoked_function_arn: str = "arn:aws:lambda:ap-northeast-1:123456789012:function:orders-api"
aws_request_id: str = "c6af9ac6-7b61-11e6-9a41-93e812345678"
@pytest.fixture(autouse=True)
def env(monkeypatch):
monkeypatch.setenv("POWERTOOLS_SERVICE_NAME", "orders")
monkeypatch.setenv("POWERTOOLS_METRICS_NAMESPACE", "Test")
def test_get_order():
import app
event = {"resource": "/orders/1", "path": "/orders/1", "httpMethod": "GET",
"headers": {}, "multiValueHeaders": {}, "queryStringParameters": None,
"requestContext": {"requestId": "req-1", "stage": "prod"},
"body": None, "isBase64Encoded": False}
res = app.lambda_handler(event, LambdaContext())
assert res["statusCode"] == 200
assert json.loads(res["body"])["item"] == "coffee"
手元では Tracer は自動で無効になります。メトリクスを出したくないテストでは POWERTOOLS_METRICS_DISABLED=true にします。Lambda のハンドラのテスト全般は、関連記事「AWS Lambda Python入門|3.14対応handler例とローカルテスト」で扱っています。
よくある質問
Q. Logger は標準の logging と何が違いますか。
A. Logger は logging.Logger を土台にしていて、logger.info() などの使い方は同じです。違いは、出力が JSON になること、Lambda の context(関数名・リクエストID・コールドスタート)と相関IDを自動で付けること、キーの追加・削除ができることです。
Q. Lambda のログ形式を JSON にする設定(ログの詳細設定)と一緒に使えますか。 A. 使えます。ドキュメントでは、Lambda の「高度なログ設定(Advanced Logging Controls)」のログレベルが、Powertools のログレベルやサンプリングに影響すると説明されています。どちらでレベルを決めるかをそろえておくと、出るはずのログが出ないことを防げます。
Q. Metrics の料金はどう考えますか。 A. EMF で出したメトリクスは CloudWatch のカスタムメトリクスとして課金され、元のログ行は CloudWatch Logs の取り込みとして課金されます。メトリクスの数は「名前空間 × メトリクス名 × ディメンションの組み合わせ」で決まるため、ディメンションの値の種類を少なく保ちます。
Q. Tracer の代わりに OpenTelemetry を使えますか。 A. Lambda では ADOT(AWS Distro for OpenTelemetry)のレイヤーで OpenTelemetry を使えます。Powertools の Tracer は X-Ray SDK を使う実装です。OpenTelemetry の基本は、関連記事「OpenTelemetry入門|Pythonで始めるトレース・メトリクス実装」で扱っています。
Q. 出たログはどうやって探しますか。
A. JSON のキーをそのまま CloudWatch Logs Insights の fields・filter に使えます。たとえば filter correlation_id = "req-abc-123" で1リクエストのログがそろいます。関連記事「CloudWatch Logs Insightsのクエリ集|エラー・遅いリクエスト」で扱っています。
まとめ
- Logger・Metrics・Tracer はハンドラにデコレーターを重ねて使い、
POWERTOOLS_SERVICE_NAMEとPOWERTOOLS_METRICS_NAMESPACEを設定する - Logger は JSON に関数名・リクエストID・
cold_start・correlation_idを付ける。append_keysは次の呼び出しに残るためclear_state=Trueを検討する - Metrics は EMF の JSON を標準出力に書く。名前空間が無いと関数が失敗する。ディメンションに ID を入れない
- Tracer は
[tracer](aws-xray-sdk)が無いとTracer()で失敗する。関数のアクティブトレースも有効にする - REST API は
APIGatewayRestResolver、HTTP API(v2)はAPIGatewayHttpResolver。enable_validation=Trueで 422 を返す入力検証ができる - Parser・Idempotency・Batch で、イベントの型付け・二重実行の防止・部分失敗をまとめて扱える
参考・出典
確認日はいずれも 2026年10月13日です。
- Powertools for AWS Lambda (Python) ホーム:https://docs.aws.amazon.com/powertools/python/latest/
- Powertools for AWS Lambda (Python)「Installation」:https://docs.aws.amazon.com/powertools/python/latest/getting-started/install/
- Powertools for AWS Lambda (Python)「Logger」:https://docs.aws.amazon.com/powertools/python/latest/core/logger/
- Powertools for AWS Lambda (Python)「Metrics」:https://docs.aws.amazon.com/powertools/python/latest/core/metrics/
- Powertools for AWS Lambda (Python)「Tracer」:https://docs.aws.amazon.com/powertools/python/latest/core/tracer/
- Powertools for AWS Lambda (Python)「REST API(Event Handler)」:https://docs.aws.amazon.com/powertools/python/latest/core/event_handler/api_gateway/
- Powertools for AWS Lambda (Python)「Parser (Pydantic)」:https://docs.aws.amazon.com/powertools/python/latest/utilities/parser/
- Powertools for AWS Lambda (Python)「Idempotency」:https://docs.aws.amazon.com/powertools/python/latest/utilities/idempotency/
- Powertools for AWS Lambda (Python)「Batch Processing」:https://docs.aws.amazon.com/powertools/python/latest/utilities/batch/
- Amazon CloudWatch ユーザーガイド「Specification: Embedded metric format」:https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html
- AWS Lambda 開発者ガイド「Visualize Lambda function invocations using AWS X-Ray」:https://docs.aws.amazon.com/lambda/latest/dg/services-xray.html
- GitHub「aws-powertools/powertools-lambda-python」:https://github.com/aws-powertools/powertools-lambda-python
- PyPI「aws-lambda-powertools」:https://pypi.org/project/aws-lambda-powertools/
本記事の内容は、ご自由にお使いください。
ご利用の際は、出典として本ページへのリンクを記載いただけますようお願いします。
(記載例)出典:株式会社RJC「Lambda Powertools Pythonの使い方|Logger・Tracer」
株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。
WE ARE HIRING
RJCで一緒に開発しながら、
成長を楽しみませんか?
RJCは、Web・モバイル・AIを活用した開発プロジェクトで、テックリードやPM・PMOも活躍するシステム開発会社です。会社を知る、待遇を確かめる、話を聞いてみる。気になるところ見てみてください!
- 127日年間休日
- 12時間平均残業時間
- 毎日ガチャ遊びココロも大切にする福利厚生。アマギフなどの賞品ラインナップ!
ほかにも、チケットレストラン、書籍読み放題、2年ごとの慰労報奨(休暇 or 金一封)、11期連続の黒字決算。