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期連続の黒字決算。

ABOUT RJC RJCがどんな会社か知る 考え方、研修、働き方、福利厚生、社員の前職まで。RJCのことが丸わかり! RJC丸わかりページへ JOB DESCRIPTION 仕事内容・待遇を見てみる 仕事内容、給与・待遇、選考の流れ。経験者も未経験も!応募前に知りたいこと、まとめました! 募集要項を見る ENTRY エントリーする エントリーは1〜2分・履歴書不要です。まずは話を聞いてみたい、という方でも歓迎です! エントリーフォームへ

RJCで一緒に開発しながら、 成長を楽しみませんか?