Python Lambda 函数

Python Lambda 函数生成器能够为现有 Python 项目添加 Lambda 函数。

该生成器会创建一个新的 Python Lambda 处理程序，并配置 AWS CDK 或 Terraform 基础设施。生成的后端使用 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
iacProvider	string	Inherit	The preferred IaC provider. By default this is inherited from your initial selection.

生成器输出

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

• 文件夹<模块名称>

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

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

• 文件夹<模块名称>

  • 文件夹<自定义路径>

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

基础设施

由于该生成器会根据您选择的 iacProvider 以基础设施即代码的形式输出，它将在 packages/common 目录下创建一个包含相关 CDK 构造体或 Terraform 模块的项目。

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

CDK

• 文件夹packages/common/constructs

  • 文件夹src

    • 文件夹app/ 针对特定项目/生成器的基础设施构造体

      • …
    • 文件夹core/ 被 app 目录构造体重用的通用构造体

      • …
    • index.ts 导出 app 目录构造体的入口文件
  • project.json 项目构建目标与配置

注意

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

Terraform

• 文件夹packages/common/terraform

  • 文件夹src

    • 文件夹app/ 针对特定项目/生成器的 Terraform 模块

      • …
    • 文件夹core/ 被 app 目录模块重用的通用模块

      • …
  • project.json 项目构建目标与配置

注意

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

生成器会根据所选 iacProvider 创建基础设施即代码（IaC）用于部署函数：

CDK

生成器创建 CDK 构造体用于部署函数，位于 packages/common/constructs 目录。

Terraform

生成器创建 Terraform 模块用于部署函数，位于 packages/common/terraform/src/app/lambda-functions/<函数名称> 目录。

实现函数

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

打包部署

生成器使用 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

注意

此操作会将整个项目打包为部署包。若要减小部署体积，建议使用 py#project 生成器将多个 Lambda 函数拆分到不同 Python 项目。

打包流程包含：

1. 使用 uv export 导出依赖到 requirements.txt
2. 使用 uv pip install 为 Lambda 目标平台 (x86_64-manylinux2014) 安装依赖

注意

若需切换至 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 事件源的目标：

注意

请确保事件源与选定的 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
}

注意

请确保事件源与选定的 eventSource 选项匹配，以保证事件在处理器函数中正确处理。