Python Lambda 関数

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 つの方法で生成できます:

このジェネレーターを実行@aws/nx-plugin:py#lambda-function

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

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

コマンドを組み立てる7

project必須

name必須

event

infra

iac

functionPath

preferInstallDependencies

true

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

オプション

ジェネレーターオプション7 オプション

project必須string: Lambda関数を追加するプロジェクト
name必須string: 追加する関数の名前
eventenumデフォルト: Any: Lambda関数に使用するオプションのイベントソースモデル
AnyAlbModelAPIGatewayProxyEventModelApiGatewayAuthorizerTokenApiGatewayAuthorizerRequestAPIGatewayProxyEventV2ModelApiGatewayAuthorizerRequestV2APIGatewayWebSocketMessageEventModelAPIGatewayWebSocketConnectEventModelAPIGatewayWebSocketDisconnectEventModelBedrockAgentEventModelCloudFormationCustomResourceCreateModelCloudFormationCustomResourceUpdateModelCloudFormationCustomResourceDeleteModelCloudWatchLogsModelDynamoDBStreamModelEventBridgeModelIoTCoreThingEventIoTCoreThingTypeEventIoTCoreThingTypeAssociationEventIoTCoreThingGroupEventIoTCoreAddOrRemoveFromThingGroupEventIoTCoreAddOrDeleteFromThingGroupEventKafkaMskEventModelKafkaSelfManagedEventModelKinesisDataStreamModelKinesisFirehoseModelKinesisFirehoseSqsModelLambdaFunctionUrlModelS3BatchOperationModelS3EventNotificationEventBridgeModelS3ModelS3ObjectLambdaEventS3SqsEventNotificationModelSesModelSnsModelSqsModelTransferFamilyAuthorizerVpcLatticeModelVpcLatticeV2Model
infraenumデフォルト: lambda: Lambda関数をデプロイするインフラストラクチャのタイプ
lambdanone
iacenumデフォルト: inherit: 優先するIaCプロバイダー。デフォルトでは初期選択から継承されます。
inheritcdkterraform
functionPathstring: 関数を追加するプロジェクト内のソースディレクトリのパス
preferInstallDependenciesbooleanデフォルト: 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> ディレクトリに配置されます。

アーキテクチャ

デプロイされた関数は以下のアーキテクチャを持ち、ログ、メトリクス、トレースは CloudWatch と X-Ray に送信されます：

Loading the diagram…

ジェネレーターは関数自体を提供します。イベントソースを呼び出すためにスタック内で接続します（例：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
@logger.inject_lambda_context
@metrics.log_metrics(capture_cold_start_metric=True)
@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()})

ハンドラーは @logger.inject_lambda_context でデコレートされているため、出力されるすべてのログレコードには以下も含まれます:

Lambda コンテキストからの function_name、function_memory_size、function_arn、function_request_id
呼び出しが新しい実行環境を初期化したかどうかを示す cold_start

トレーシング

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 ...

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

すべての呼び出しで追加される InvocationCount
ハンドラーの try/except によって追加される SuccessCount と ErrorCount
呼び出しが新しい実行環境を初期化したときに @metrics.log_metrics(capture_cold_start_metric=True) によって出力される ColdStart

型安全性

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
}