Python Lambda 函数

Python Lambda Function 生成器提供了向现有 Python 项目添加 lambda 函数的能力。

此生成器创建一个新的 Python lambda 处理程序，并配置 AWS CDK 或 Terraform 基础设施。生成的后端使用 AWS Lambda 进行无服务器部署，可选择使用 AWS Lambda Powertools 的 Parser 实现类型安全。它设置了 AWS Lambda Powertools 以实现可观测性，包括日志记录、AWS X-Ray 追踪和 Cloudwatch 指标。

用法

生成 Lambda 函数

您可以通过两种方式生成新的 Lambda 函数：

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

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

选项

参数	类型	默认值	描述
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 项目图，仍会运行安装）；在最后统一安装一次。

生成器输出

生成器将向您的项目添加以下文件：

• 文件夹<module-name>

  • <lambda-function>.py Function implementation

如果提供了 functionPath 选项，生成器将在指定路径中添加必要的文件：

• 文件夹<module-name>

  • 文件夹<custom-path>

    • <function-name>.py Function implementation

基础设施

由于此生成器根据您选择的 iac 提供基础设施即代码，它将在 packages/common 中创建一个项目，其中包含相关的 CDK 构造或 Terraform 模块。

通用基础设施即代码项目的结构如下：

CDK

• 文件夹packages/common/constructs

  • 文件夹src

    • 文件夹app/ Constructs for infrastructure specific to a project/generator

      • …
    • 文件夹core/ Generic constructs which are reused by constructs in app

      • …
    • index.ts Entry point exporting constructs from app
  • project.json Project build targets and configuration

Generated Project

此项目使用 ts#project 生成器生成，因此配置了相同的构建目标。

Terraform

• 文件夹packages/common/terraform

  • 文件夹src

    • 文件夹app/ Terraform modules for infrastructure specific to a project/generator

      • …
    • 文件夹core/ Generic modules which are reused by modules in app

      • …
  • project.json Project build targets and configuration

Generated Project

此项目使用 terraform#project 生成器生成，因此配置了相同的构建目标。

生成器根据您选择的 iac 创建用于部署函数的基础设施即代码：

CDK

生成器创建可用于部署函数的 CDK 构造，这些构造位于 packages/common/constructs 目录中。

Terraform

生成器创建可用于部署函数的 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

生成器会自动设置以下几个功能：

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()})

关联 ID

建议为所有唯一请求设置关联 ID，以便更轻松地进行调试和监控。请参阅 aws powertools logger 文档了解关联 ID 的最佳实践。

日志记录器自动包含：

• 事件请求
• 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 ...

默认指标包括：

• 调用次数
• 成功/失败次数
• 冷启动指标

类型安全

event ≠ Any

如果您在生成 lambda 函数时选择了 event，您的函数将使用 @event_parser from AWS Lambda Powertools 进行检测。例如：

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

这允许您使用 Pydantic 定义数据模型，其方式类似于使用 Fast API。

自定义事件数据

如果您在事件中嵌套了自定义数据，例如 DynamoDB 流或 EventBridge 事件，您可能会受益于使用 Envelopes 为该自定义数据提供类型安全。

event = Any

如果您为 event 选择了 Any，则不会应用 @event_parser 装饰器，处理程序将接收未类型化的事件字典。如果您想要 Pydantic 验证和类型化字段访问，请使用特定的 event 重新生成。

打包

生成器使用 uv 自动配置 Lambda 部署包的 Python 打包：

pnpm

pnpm nx bundle <project-name>

yarn

yarn nx bundle <project-name>

npm

npx nx bundle <project-name>

bun

bunx nx bundle <project-name>

打包行为

这会将整个项目打包为部署包。为了最小化部署大小，您可能希望使用 py#project 生成器生成的多个 Python 项目来拆分多个 lambda 函数。

此过程使用：

1. uv export 将您的 Python 依赖项导出到 requirements.txt 文件
2. uv pip install 为目标平台（x86_64-manylinux_2_28）安装依赖项以进行 Lambda 部署

架构更改

如果您将 lambda 架构更改为 ARM，您需要编辑打包目标以使用 aarch64-manylinux_2_28 目标。

部署您的函数

此生成器会根据您选择的 iac 创建 CDK 或 Terraform 基础设施即代码。您可以使用它来部署您的函数。

CDK

此生成器会在 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');
  }
}

这将设置：

1. AWS Lambda 函数
2. CloudWatch 日志组
3. X-Ray 跟踪配置
4. CloudWatch 指标命名空间

然后，此函数可以用作任何 lambda 事件源的目标：

事件源匹配

确保事件源与所选的 event 选项匹配，以确保事件在处理程序函数中得到正确处理。

下面的示例演示了使用 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)],
    });
  }
}

Terraform

此生成器会在 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
  ]
}

这将设置：

1. AWS Lambda 函数
2. CloudWatch 日志组
3. X-Ray 跟踪配置
4. 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
}

事件源匹配

确保事件源与所选的 event 选项匹配，以确保事件在处理程序函数中得到正确处理。