Pythonのラムダ関数

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

このジェネレータは、AWS CDKまたはTerraformのインフラストラクチャ設定を含む新しい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
iacProvider	string	Inherit	The preferred IaC provider. By default this is inherited from your initial selection.

ジェネレータの出力

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

• Directory<module-name>

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

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

• Directory<module-name>

  • Directory<custom-path>

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

インフラストラクチャ

このジェネレータは選択した iacProvider に基づいてInfrastructure as Codeを生成するため、packages/common に関連するCDKコンストラクトまたはTerraformモジュールを含むプロジェクトを作成します。

共通のInfrastructure as Codeプロジェクトは以下の構造を持ちます：

CDK

• Directorypackages/common/constructs

  • Directorysrc

    • Directoryapp/ プロジェクト/ジェネレータ固有のインフラストラクチャ用コンストラクト

      • …
    • Directorycore/ app 内のコンストラクトで再利用される汎用コンストラクト

      • …
    • index.ts app からコンストラクトをエクスポートするエントリーポイント
  • project.json プロジェクトのビルドターゲットと設定

Note

このプロジェクトは ts#project ジェネレータを使用して生成されているため、同じビルドターゲットを設定します。

Terraform

• Directorypackages/common/terraform

  • Directorysrc

    • Directoryapp/ プロジェクト/ジェネレータ固有のインフラストラクチャ用Terraformモジュール

      • …
    • Directorycore/ app 内のモジュールで再利用される汎用モジュール

      • …
  • project.json プロジェクトのビルドターゲットと設定

Note

このプロジェクトは terraform#project ジェネレータを使用して生成されているため、同じビルドターゲットを設定します。

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

CDK

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

Terraform

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

関数の実装

メインの関数実装は<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を選択できます。

バンドリング

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

pnpm

pnpm nx run <project-name>:bundle

yarn

yarn nx run <project-name>:bundle

npm

npx nx run <project-name>:bundle

bun

bunx nx run <project-name>:bundle

Note

デプロイメントサイズを最小化するため、複数のLambda関数をpy#projectジェネレータで生成した複数のPythonプロジェクトに分割することを検討してください。

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

1. uv exportでPython依存関係をrequirements.txtにエクスポート
2. uv pip installでLambdaデプロイメント用ターゲットプラットフォーム（x86_64-manylinux2014）に依存関係をインストール

Note

LambdaアーキテクチャをARMに変更する場合、バンドルターゲットをaarch64-manylinux2014に変更する必要があります。

関数のデプロイ

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

CDK

このジェネレータは 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 オプションと一致していることを確認してください。

以下の例は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) {
    // スタックに関数を追加
    const fn = new MyProjectMyFunction(this, 'MyFunction');

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

Terraform

このジェネレータは common/terraform フォルダに関数をデプロイするためのTerraformモジュールを作成します。Terraform設定で以下のように使用できます：

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

  # Lambda関数の環境変数
  env = {
    SOME_VARIABLE = "some value"
  }

  # 追加のIAMポリシーが必要な場合
  additional_iam_policy_statements = [
    # 関数に必要な追加の権限を記述
  ]
}

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

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

この関数は任意のLambdaイベントソースのターゲットとして使用できます。以下の例はEventBridgeを使用してスケジュールでLambda関数を起動するTerraformコードを示しています：

# スケジュール実行用EventBridgeルール
resource "aws_cloudwatch_event_rule" "my_function_schedule" {
  name                = "my-function-schedule"
  description         = "15分毎に関数を起動"
  schedule_expression = "cron(15 * * * ? *)"
}

# EventBridgeターゲット
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
}

# EventBridgeがLambda関数を起動するための権限
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
}

Note

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