Python Lambda 関数

Filter this guidePick generator option values to hide sections that don't apply.

event

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

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

使用方法

Lambda 関数の生成

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

CLI
VSCode

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

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

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

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

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

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

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

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

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

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

オプション

パラメータ	型	デフォルト	説明
project 必須	string	-	Lambda関数を追加するプロジェクト
name 必須	string	-	追加する関数の名前
functionPath	string	-	関数を追加するプロジェクト内のソースディレクトリのパス
event	Any \| AlbModel \| APIGatewayProxyEventModel \| ApiGatewayAuthorizerToken \| ApiGatewayAuthorizerRequest \| APIGatewayProxyEventV2Model \| ApiGatewayAuthorizerRequestV2 \| APIGatewayWebSocketMessageEventModel \| APIGatewayWebSocketConnectEventModel \| APIGatewayWebSocketDisconnectEventModel \| BedrockAgentEventModel \| CloudFormationCustomResourceCreateModel \| CloudFormationCustomResourceUpdateModel \| CloudFormationCustomResourceDeleteModel \| CloudWatchLogsModel \| DynamoDBStreamModel \| EventBridgeModel \| IoTCoreThingEvent \| IoTCoreThingTypeEvent \| IoTCoreThingTypeAssociationEvent \| IoTCoreThingGroupEvent \| IoTCoreAddOrRemoveFromThingGroupEvent \| IoTCoreAddOrDeleteFromThingGroupEvent \| KafkaMskEventModel \| KafkaSelfManagedEventModel \| KinesisDataStreamModel \| KinesisFirehoseModel \| KinesisFirehoseSqsModel \| LambdaFunctionUrlModel \| S3BatchOperationModel \| S3EventNotificationEventBridgeModel \| S3Model \| S3ObjectLambdaEvent \| S3SqsEventNotificationModel \| SesModel \| SnsModel \| SqsModel \| TransferFamilyAuthorizer \| VpcLatticeModel \| VpcLatticeV2Model	Any	Lambda関数に使用するオプションのイベントソースモデル
infra	lambda \| none	lambda	Lambda関数をデプロイするインフラストラクチャのタイプ
iac	inherit \| cdk \| terraform	inherit	優先するIaCプロバイダー。デフォルトでは初期選択から継承されます。
preferInstallDependencies	boolean	true	ジェネレーター実行後に依存関係のインストールを優先するかどうか。複数のジェネレーターをバッチ処理する際にインストールを延期する場合はfalseに設定します（後続のジェネレーターがNxプロジェクトグラフを計算できるよう、必要に応じてインストールは実行されます）。最後に一度だけインストールします。

ジェネレーターの出力

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

Directory<module-name>
- <lambda-function>.py Function implementation

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

Directory<module-name>
- Directory<custom-path>
  - <function-name>.py Function implementation

インフラストラクチャ

このジェネレーターは、選択した iac に基づいてインフラストラクチャをコードとして提供するため、関連する CDK コンストラクトまたは Terraform モジュールを含む packages/common にプロジェクトを作成します。

共通のインフラストラクチャコードプロジェクトは、次のように構成されています：

CDK
Terraform

Directorypackages/common/constructs
- Directorysrc
  - Directoryapp/ プロジェクト/ジェネレーター固有のインフラストラクチャ用のコンストラクト
    …
  - Directorycore/ app 内のコンストラクトによって再利用される汎用コンストラクト
    …
  - index.ts app からコンストラクトをエクスポートするエントリーポイント
- project.json プロジェクトのビルドターゲットと設定

Directorypackages/common/terraform
- Directorysrc
  - Directoryapp/ プロジェクト/ジェネレーター固有のインフラストラクチャ用の Terraform モジュール
    …
  - Directorycore/ app 内のモジュールによって再利用される汎用モジュール
    …
- project.json プロジェクトのビルドターゲットと設定

ジェネレーターは、選択した iac に基づいて関数をデプロイするための Infrastructure as Code を作成します:

CDK
Terraform

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

ジェネレーターは、関数をデプロイするために使用できる Terraform モジュールを作成します。これは packages/common/terraform/src/app/lambda-functions/<function-name> ディレクトリに配置されます。

アーキテクチャ

デプロイされた関数は以下のアーキテクチャを持ちます：

ジェネレーターは関数自体を提供します。イベントソースを呼び出すためにスタック内で接続します（例：API Gateway 統合、EventBridge ルール、S3 通知、または SQS イベントソースマッピング）。

関数の実装

メインの関数実装は <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: Implement
        metrics.add_metric(name="SuccessCount", unit=MetricUnit.Count, value=1)
        # TODO: Implement success response if required
    except Exception as e:
        logger.exception(e)
        metrics.add_metric(name="ErrorCount", unit=MetricUnit.Count, value=1)
        # TODO: Implement error response if required

ジェネレーターは、いくつかの機能を自動的にセットアップします:

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

AWS Lambda Powertools による可観測性

ログ記録

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

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

ロガーは自動的に以下を含みます:

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

トレーシング

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

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    # Creates a new subsegment
    with tracer.provider.in_subsegment("function-subsegment"):
        # Your logic here
        return ....

メトリクス

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

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

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

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

型安全性

Lambda 関数を生成する際に event を選択した場合、関数は AWS Lambda Powertools の @event_parser でインストルメント化されます。例えば:

@event_parser(model=EventBridgeModel)
def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    event.detail_type # <- type-safe with IDE autocompletion

これにより、Fast API での作業と同様の方法で、Pydantic を使用してデータモデルを定義できます。

event に Any を選択した場合、@event_parser デコレーターは適用されず、ハンドラーは型なしのイベント dict を受け取ります。Pydantic バリデーションと型付きフィールドアクセスが必要な場合は、特定の event で再生成してください。

バンドリング

ジェネレーターは、uv を使用して Lambda デプロイメントパッケージの Python バンドリングを自動的に設定します:

pnpm nx bundle <project-name>

yarn nx bundle <project-name>

npx nx bundle <project-name>

bunx nx bundle <project-name>

このプロセスでは以下を使用します:

uv export で Python 依存関係を requirements.txt ファイルにエクスポート
uv pip install で Lambda デプロイメント用のターゲットプラットフォーム (x86_64-manylinux_2_28) の依存関係をインストール

関数のデプロイ

このジェネレーターは、選択した iac に基づいて CDK または Terraform のインフラストラクチャコードを作成します。これを使用して関数をデプロイできます。

CDK
Terraform

このジェネレーターは、common/constructs フォルダに関数をデプロイするための CDK コンストラクトを作成します。これを CDK アプリケーションで使用できます：

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

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    // Add the function to your stack
    const fn = new MyProjectMyFunction(this, 'MyFunction');
  }
}

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

AWS Lambda 関数
CloudWatch ロググループ
X-Ray トレース設定
CloudWatch メトリクス名前空間

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

以下の例は、EventBridge を使用してスケジュールに従って Lambda 関数を呼び出すための CDK コードを示しています：

import { 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) {
    // Add the function to your stack
    const fn = new MyProjectMyFunction(this, 'MyFunction');

    // Add the function to an EventBridge scheduled rule
    const eventRule = new Rule(this, 'MyFunctionScheduleRule', {
      schedule: Schedule.cron({ minute: '15' }),
      targets: [new LambdaFunction(fn)],
    });
  }
}

このジェネレーターは、common/terraform フォルダに関数をデプロイするための Terraform モジュールを作成します。デプロイごとに共有 core/asset-bucket モジュールを一度インスタンス化し、その bucket_name を asset_bucket_name 入力に渡します：

module "asset_bucket" {
  source = "../../common/terraform/src/core/asset-bucket"
}

module "my_project_my_function" {
  source = "../../common/terraform/src/app/lambda-functions/my-project-my-function"

  asset_bucket_name = module.asset_bucket.bucket_name

  env = {
    SOME_VARIABLE = "some value"
  }

  additional_iam_policy_statements = [
    # Add any additional permissions your function needs
  ]
}

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

AWS Lambda 関数
CloudWatch ロググループ
X-Ray トレース設定
CloudWatch メトリクス名前空間

この関数は、任意の Lambda イベントソースのターゲットとして使用できます。以下の例は、EventBridge を使用してスケジュールに従って Lambda 関数を呼び出すための Terraform コードを示しています：

# EventBridge rule for scheduled execution
resource "aws_cloudwatch_event_rule" "my_function_schedule" {
  name                = "my-function-schedule"
  description         = "Trigger my function every 15 minutes"
  schedule_expression = "cron(15 * * * ? *)"
}

# EventBridge target
resource "aws_cloudwatch_event_target" "lambda_target" {
  rule      = aws_cloudwatch_event_rule.my_function_schedule.name
  target_id = "MyFunctionTarget"
  arn       = module.my_project_my_function.function_arn
}

# Permission for EventBridge to invoke the Lambda function
resource "aws_lambda_permission" "allow_eventbridge" {
  statement_id  = "AllowExecutionFromEventBridge"
  action        = "lambda:InvokeFunction"
  function_name = module.my_project_my_function.function_name
  principal     = "events.amazonaws.com"
  source_arn    = aws_cloudwatch_event_rule.my_function_schedule.arn
}