Python Lambda 函数

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

该生成器将创建带有 AWS CDK 基础设施配置的新 Python Lambda 处理器。生成的后端使用 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-function>.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: 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 以便调试和监控，请参考 AWS Powertools 日志文档 获取最佳实践。

日志记录器自动包含：

• 事件请求
• 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 定义数据模型，类似 FastAPI 的工作方式。

提示

如果事件中存在嵌套的自定义数据（如 DynamoDB 流或 EventBridge 事件），建议使用 Envelopes 为这些数据提供类型安全。

若不需要事件类型校验，可为 eventSource 选择 Any 选项。

部署函数

Python Lambda Function 生成器会在 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)],
    });
  }
}