Pythonのラムダ関数

Python Lambda Function ジェネレータは、既存のPythonプロジェクトにLambda関数を追加する機能を提供します。

このジェネレータはAWS CDKインフラストラクチャ設定を含む新しいPython Lambdaハンドラを作成します。生成されるバックエンドはサーバーレスデプロイにAWS Lambdaを使用し、AWS Lambda PowertoolsのParserを使った型安全性をオプションで追加します。ロギング、AWS X-Rayトレーシング、Cloudwatchメトリクスを含む可観測性のためにAWS Lambda Powertoolsを設定します。

使用方法

Lambda関数の生成

新しいLambda関数は2つの方法で生成できます:

VSCode

1. インストール Nx Console VSCode Plugin まだインストールしていない場合
2. VSCodeでNxコンソールを開く
3. クリック Generate (UI) "Common Nx Commands"セクションで
4. 検索 @aws/nx-plugin - py#lambda-function
5. 必須パラメータを入力
6. クリック Generate

CLI

pnpm

pnpm nx g @aws/nx-plugin:py#lambda-function

yarn

yarn nx g @aws/nx-plugin:py#lambda-function

npm

npx nx g @aws/nx-plugin:py#lambda-function

bun

bunx nx g @aws/nx-plugin:py#lambda-function

変更されるファイルを確認するためにドライランを実行することもできます

pnpm

pnpm nx g @aws/nx-plugin:py#lambda-function --dry-run

yarn

yarn nx g @aws/nx-plugin:py#lambda-function --dry-run

npm

npx nx g @aws/nx-plugin:py#lambda-function --dry-run

bun

bunx nx g @aws/nx-plugin:py#lambda-function --dry-run

オプション

パラメータ	型	デフォルト	説明
project 必須	string	-	The project to add the lambda function to
functionName 必須	string	-	The name of the function to add
functionPath	string	-	The path within the project source directory to add the function to
eventSource 必須	string	Any	Optional event source model to use for the lambda function

ジェネレータの出力

ジェネレータはプロジェクトに以下のファイルを追加します:

• Directory<module-name>

  • <lambda-function>.py 関数実装

ジェネレータはまた、関数のデプロイに使用できるCDKコンストラクトをpackages/common/constructsディレクトリに作成します。

functionPathオプションが指定された場合、ジェネレータは指定されたパスに必要なファイルを追加します:

• Directory<module-name>

  • Directory<custom-path>

    • <function-name>.py 関数実装

関数の実装

メインの関数実装は<function-name>.pyにあります。以下は実装例です:

import os

from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.utilities.parser import event_parser
from aws_lambda_powertools.utilities.parser.models import EventBridgeModel
from aws_lambda_powertools.utilities.typing import LambdaContext

os.environ["POWERTOOLS_METRICS_NAMESPACE"] = "Foo"
os.environ["POWERTOOLS_SERVICE_NAME"] = "Foo"

logger: Logger = Logger()
metrics: Metrics = Metrics()
tracer: Tracer = Tracer()

@tracer.capture_lambda_handler
@metrics.log_metrics
@event_parser(model=EventBridgeModel)
def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    logger.info("Received event", extra={"event": event.model_dump() })
    metrics.add_metric(name="InvocationCount", unit=MetricUnit.Count, value=1)

    try:
        # TODO: 実装
        metrics.add_metric(name="SuccessCount", unit=MetricUnit.Count, value=1)
        # TODO: 必要に応じて成功レスポンスを実装
    except Exception as e:
        logger.exception(e)
        metrics.add_metric(name="ErrorCount", unit=MetricUnit.Count, value=1)
        # TODO: 必要に応じてエラーレスポンスを実装

ジェネレータは以下の機能を自動的に設定します:

1. 可観測性のためのAWS Lambda Powertools統合
2. メトリクス収集
3. @event_parserを使用した型安全性

AWS Lambda Powertoolsによる可観測性

ロギング

ジェネレータはAWS Lambda Powertoolsを使用した構造化ロギングを設定します。

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    logger.info("Received event", extra={"event": event.model_dump()})

Tip

デバッグと監視を容易にするため、すべてのユニークなリクエストに相関IDを設定することを推奨します。相関IDのベストプラクティスについてはaws powertools loggerドキュメントを参照してください。

ロガーには自動的に以下が含まれます:

• イベントリクエスト
• Lambdaコンテキスト情報
• コールドスタートインジケータ

トレーシング

AWS X-Rayトレーシングが自動的に設定されます。カスタムサブセグメントを追加できます:

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    # 新しいサブセグメントを作成
    with tracer.provider.in_subsegment("function-subsegment"):
        # ここにロジックを実装
        return ....

メトリクス

CloudWatchメトリクスは各リクエストに対して自動的に収集されます。カスタムメトリクスを追加できます:

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    metrics.add_metric(name="NewMetric", unit=MetricUnit.Count, value=1)
    return ...

デフォルトのメトリクスには以下が含まれます:

• 呼び出し回数
• 成功/失敗回数
• コールドスタートメトリクス

型安全性

Lambda関数生成時にeventSourceを選択した場合、関数はAWS Lambda Powertoolsの@event_parserで装飾されます。例:

@event_parser(model=EventBridgeModel)
def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    event.detail_type # <- IDEのオートコンプリートが効く型安全なアクセス

これはFast APIと同様の方法でPydanticを使用したデータモデル定義を可能にします。

Tip

DynamoDBストリームやEventBridgeイベント内のカスタムデータなど、ネストされたカスタムデータがある場合、Envelopesを使用して型安全性を確保できます。

イベントの型指定が必要ない場合は、eventSourceでAnyを選択できます。

関数のデプロイ

Python Lambda Functionジェネレータはcommon/constructsフォルダにCDKコンストラクトを作成します。CDKアプリケーションで使用できます:

import { MyProjectMyFunction } from ':my-scope/common-constructs';

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    // スタックに関数を追加
    const fn = new MyProjectMyFunction(this, 'MyFunction');
  }
}

これにより以下が設定されます:

1. AWS Lambda関数
2. CloudWatchロググループ
3. X-Rayトレーシング設定
4. CloudWatchメトリクスネームスペース

この関数は任意のLambdaイベントソースのターゲットとして使用できます:

Note

イベントがハンドラ関数内で適切に処理されるよう、イベントソースが選択したeventSourceオプションと一致していることを確認してください。

以下の例はEvent Bridgeを使用したスケジュール実行でLambda関数を呼び出すCDKコードを示しています:

import { EventPattern, Rule, Schedule } from 'aws-cdk-lib/aws-events';
import { LambdaFunction } from 'aws-cdk-lib/aws-events-targets';
import { MyProjectMyFunction } from ':my-scope/common-constructs';

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    // スタックに関数を追加
    const fn = new MyProjectMyFunction(this, 'MyFunction');

    // EventBridgeタスクに関数を追加
    const eventRule = new Rule(this, 'MyFunctionScheduleRule', {
      schedule: Schedule.cron({ minute: '15' }),
      targets: [new LambdaFunction(fn)],
    });
  }
}