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 函数：

运行此生成器@aws/nx-plugin:py#lambda-function

CLI
VSCode

$ pnpm nx g @aws/nx-plugin:py#lambda-function

$ yarn nx g @aws/nx-plugin:py#lambda-function

$ npx nx g @aws/nx-plugin:py#lambda-function

$ bunx nx g @aws/nx-plugin:py#lambda-function

安装 Nx Console VSCode Plugin 如果您尚未安装
在VSCode中打开Nx控制台
点击 Generate (UI) 在"Common Nx Commands"部分
搜索 @aws/nx-plugin - py#lambda-function
填写必需参数
点击 Generate

构建你的命令7

project必需

name必需

event

infra

iac

functionPath

preferInstallDependencies

true

--dry-run您还可以执行试运行以查看哪些文件会被更改

选项

生成器选项7 个选项

project必需string: 要添加 Lambda 函数的项目
name必需string: 要添加的函数名称
eventenum默认值: Any: 用于 lambda 函数的可选事件源模型
AnyAlbModelAPIGatewayProxyEventModelApiGatewayAuthorizerTokenApiGatewayAuthorizerRequestAPIGatewayProxyEventV2ModelApiGatewayAuthorizerRequestV2APIGatewayWebSocketMessageEventModelAPIGatewayWebSocketConnectEventModelAPIGatewayWebSocketDisconnectEventModelBedrockAgentEventModelCloudFormationCustomResourceCreateModelCloudFormationCustomResourceUpdateModelCloudFormationCustomResourceDeleteModelCloudWatchLogsModelDynamoDBStreamModelEventBridgeModelIoTCoreThingEventIoTCoreThingTypeEventIoTCoreThingTypeAssociationEventIoTCoreThingGroupEventIoTCoreAddOrRemoveFromThingGroupEventIoTCoreAddOrDeleteFromThingGroupEventKafkaMskEventModelKafkaSelfManagedEventModelKinesisDataStreamModelKinesisFirehoseModelKinesisFirehoseSqsModelLambdaFunctionUrlModelS3BatchOperationModelS3EventNotificationEventBridgeModelS3ModelS3ObjectLambdaEventS3SqsEventNotificationModelSesModelSnsModelSqsModelTransferFamilyAuthorizerVpcLatticeModelVpcLatticeV2Model
infraenum默认值: lambda: 用于部署 Lambda 函数的基础设施类型。
lambdanone
iacenum默认值: inherit: 首选的 IaC 提供商。默认情况下,这继承自您的初始选择。
inheritcdkterraform
functionPathstring: 在项目源代码目录中添加函数的路径
preferInstallDependenciesboolean默认值: 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
Terraform

文件夹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

文件夹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

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

CDK
Terraform

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

生成器创建可用于部署函数的 Terraform 模块，该模块位于 packages/common/terraform/src/app/lambda-functions/<function-name> 目录中。

架构

已部署的函数具有以下架构，其日志、指标和跟踪信息将发送到 CloudWatch 和 X-Ray：

Loading the diagram…

生成器提供函数本身；您需要在堆栈中连接事件源以调用它（例如，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
@logger.inject_lambda_context
@metrics.log_metrics(capture_cold_start_metric=True)
@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

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

AWS Lambda Powertools 集成以实现可观测性
指标收集
使用 @event_parser 实现类型安全

使用 AWS Lambda Powertools 实现可观测性

日志记录

生成器使用 AWS Lambda Powertools 配置结构化日志记录。

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    logger.info("Received event", extra={"event": event.model_dump()})

处理程序使用 @logger.inject_lambda_context 装饰，因此它发出的每条日志记录还包含：

来自 Lambda 上下文的 function_name、function_memory_size、function_arn 和 function_request_id
cold_start，指示调用是否初始化了新的执行环境

追踪

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 ...

默认指标包括：

InvocationCount，在每次调用时添加
SuccessCount 和 ErrorCount，由处理程序的 try/except 添加
ColdStart，由 @metrics.log_metrics(capture_cold_start_metric=True) 在调用初始化新执行环境时发出

类型安全

如果您在生成 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。

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

打包

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

pnpm nx bundle <project-name>

yarn nx bundle <project-name>

npx nx bundle <project-name>

bunx nx bundle <project-name>

此过程使用：

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

部署您的函数

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

CDK
Terraform

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

这将设置：

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

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

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

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

这将设置：

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