Python Lambda 函数

Python Lambda 函数生成器支持在现有 Python 项目中添加 Lambda 函数。

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

使用方法

生成 Lambda 函数

有两种方式可以生成新的 Lambda 函数：

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

生成器输出

生成器将在项目中添加以下文件：

• 文件夹<模块名称>

  • <lambda函数>.py 函数实现

生成器还会在 packages/common/constructs 目录下创建用于部署函数的 CDK 构造。

若提供了 functionPath 选项，生成器会将必要文件添加到指定路径：

• 文件夹<模块名称>

  • 文件夹<自定义路径>

    • <函数名称>.py 函数实现

实现函数

主函数实现位于 <函数名称>.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()})

提示

建议为所有唯一请求设置关联 ID 以便调试和监控。请参考 AWS Powertools 日志文档 了解关联 ID 最佳实践。

日志记录器自动包含：

• 事件请求
• 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 自动补全

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

提示

若事件中包含自定义数据（如 DynamoDB 流或 EventBridge 事件），使用 Envelopes 可为该数据提供类型安全。

若不需要事件类型定义，可在 eventSource 选项中选择 Any。

部署函数

Python Lambda 函数生成器在 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 事件源 的目标：

注意

请确保事件源与所选 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)],
    });
  }
}