Funciones Lambda de Python

El generador de Funciones Lambda de Python proporciona la capacidad de agregar una función lambda a un proyecto de Python existente.

Este generador crea un nuevo manejador de lambda de Python con configuración de infraestructura AWS CDK o Terraform. El backend generado utiliza AWS Lambda para despliegue serverless, con seguridad de tipos opcional usando el Parser de AWS Lambda Powertools. Configura AWS Lambda Powertools para observabilidad, incluyendo registro, rastreo de AWS X-Ray y métricas de Cloudwatch.

Uso

Generar una Función Lambda

Puedes generar una nueva Función Lambda de dos maneras:

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

También puede realizar una ejecución en seco para ver qué archivos se cambiarían

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. Instale el Nx Console VSCode Plugin si aún no lo ha hecho
2. Abra la consola Nx en VSCode
3. Haga clic en Generate (UI) en la sección "Common Nx Commands"
4. Busque @aws/nx-plugin - py#lambda-function
5. Complete los parámetros requeridos
6. Haga clic en Generate

Opciones

Parámetro	Tipo	Predeterminado	Descripción
project Requerido	string	-	El proyecto al que agregar la función lambda
name Requerido	string	-	El nombre de la función a agregar
functionPath	string	-	La ruta dentro del directorio fuente del proyecto donde agregar la función
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	Modelo de origen de evento opcional para usar con la función lambda
infra	lambda | none	lambda	El tipo de infraestructura con la que desplegar tu función Lambda.
iac	inherit | cdk | terraform	inherit	El proveedor de IaC preferido. Por defecto, se hereda de tu selección inicial.
preferInstallDependencies	boolean	true	Si se prefiere instalar las dependencias después de que se ejecute el generador. Establecer en false para diferir la instalación al procesar múltiples generadores en lote (la instalación aún se ejecuta si es necesario para que los generadores subsecuentes puedan calcular el grafo de proyectos de Nx); instalar una vez al final.

Salida del Generador

El generador agregará los siguientes archivos a tu proyecto:

• Directorio<module-name>

  • <lambda-function>.py Function implementation

Si se proporciona la opción functionPath, el generador agregará los archivos necesarios a la ruta especificada:

• Directorio<module-name>

  • Directorio<custom-path>

    • <function-name>.py Function implementation

Infraestructura

Dado que este generador proporciona infraestructura como código basada en tu iac elegido, creará un proyecto en packages/common que incluye las construcciones CDK o módulos Terraform relevantes.

El proyecto común de infraestructura como código está estructurado de la siguiente manera:

CDK

• Directoriopackages/common/constructs

  • Directoriosrc

    • Directorioapp/ Constructs for infrastructure specific to a project/generator

      • …
    • Directoriocore/ Generic constructs which are reused by constructs in app

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

Proyecto Generado

Este proyecto se genera utilizando el generador ts#project y por lo tanto configura los mismos objetivos de compilación.

Terraform

• Directoriopackages/common/terraform

  • Directoriosrc

    • Directorioapp/ Terraform modules for infrastructure specific to a project/generator

      • …
    • Directoriocore/ Generic modules which are reused by modules in app

      • …
  • project.json Project build targets and configuration

Proyecto Generado

Este proyecto se genera utilizando el generador terraform#project y por lo tanto configura los mismos objetivos de compilación.

El generador crea infraestructura como código para desplegar tu función basándose en tu iac seleccionado:

CDK

El generador crea construcciones CDK que pueden usarse para desplegar tu función, las cuales residen en el directorio packages/common/constructs.

Terraform

El generador crea un módulo de Terraform que puede usarse para desplegar tu función, el cual reside en el directorio packages/common/terraform/src/app/lambda-functions/<function-name>.

Arquitectura

La función desplegada tiene la siguiente arquitectura:

El generador proporciona la función en sí; usted conecta la fuente de eventos en su stack para invocarla (por ejemplo, una integración de API Gateway, una regla de EventBridge, una notificación de S3 o un mapeo de fuente de eventos de SQS).

Implementar tu Función

La implementación principal de la función está en <function-name>.py. Aquí hay un ejemplo:

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

El generador configura varias características automáticamente:

1. Integración de AWS Lambda Powertools para observabilidad
2. Recolección de métricas
3. Seguridad de tipos usando @event_parser

Observabilidad con AWS Lambda Powertools

Registro

El generador configura el registro estructurado usando AWS Lambda Powertools.

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

ID de Correlación

Se recomienda establecer el ID de correlación para todas las solicitudes únicas para facilitar la depuración y el monitoreo. Por favor, consulta la documentación del logger de aws powertools para las mejores prácticas de ID de correlación.

El logger incluye automáticamente:

• Solicitudes de eventos
• Información del contexto de Lambda
• Indicadores de arranque en frío

Rastreo

El rastreo de AWS X-Ray se configura automáticamente. Puedes agregar subsegmentos personalizados a tus rastros:

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    # Creates a new subsegment
    with tracer.provider.in_subsegment("function-subsegment"):
        # Your logic here
        return ....

Métricas

Las métricas de CloudWatch se recopilan automáticamente para cada solicitud. Puedes agregar métricas personalizadas:

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    metrics.add_metric(name="NewMetric", unit=MetricUnit.Count, value=1)
    return ...

Las métricas predeterminadas incluyen:

• Conteos de invocaciones
• Conteos de éxito/fallo
• Métricas de arranque en frío

Seguridad de Tipos

event ≠ Any

Si elegiste un event al generar tu función lambda, tu función está instrumentada con @event_parser de AWS Lambda Powertools. Por ejemplo:

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

Esto te permite definir modelos de datos usando Pydantic, de manera similar a trabajar con Fast API.

Datos de Evento Personalizados

Si tienes datos personalizados anidados dentro de un evento, por ejemplo un stream de DynamoDB o un evento de EventBridge, puedes beneficiarte de usar Envelopes para proporcionar seguridad de tipos para esos datos personalizados.

event = Any

Si seleccionaste Any para tu event, el decorador @event_parser no se aplica y el manejador recibe un diccionario de evento sin tipo. Regenera con un event específico si deseas validación de Pydantic y acceso a campos con tipo.

Empaquetado

El generador configura automáticamente el empaquetado de Python para paquetes de despliegue de Lambda usando uv:

pnpm

pnpm nx bundle <project-name>

yarn

yarn nx bundle <project-name>

npm

npx nx bundle <project-name>

bun

bunx nx bundle <project-name>

Comportamiento del Empaquetado

Esto empaqueta todo el proyecto como el paquete de despliegue. Para minimizar el tamaño del despliegue, es posible que desees dividir múltiples funciones lambda en múltiples proyectos de Python generados con el generador py#project.

Este proceso utiliza:

1. uv export para exportar tus dependencias de Python a un archivo requirements.txt
2. uv pip install para instalar dependencias para la plataforma de destino (x86_64-manylinux_2_28) para el despliegue de Lambda

Cambios de Arquitectura

Si cambias tu arquitectura lambda a ARM, necesitarás editar tu objetivo de empaquetado para usar el objetivo aarch64-manylinux_2_28.

Desplegar tu Función

Este generador crea infraestructura como código CDK o Terraform basada en tu iac seleccionado. Puedes usar esto para desplegar tu función.

CDK

Este generador crea un construct CDK para desplegar tu función en la carpeta common/constructs. Puedes usar esto en una aplicación 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');
  }
}

Esto configura:

1. Función AWS Lambda
2. Grupo de logs de CloudWatch
3. Configuración de rastreo X-Ray
4. Espacio de nombres de métricas de CloudWatch

Esta función puede entonces ser usada como objetivo para cualquier fuente de eventos de lambda:

Coincidencia de Fuente de Eventos

Asegúrate de que la fuente de eventos coincida con la opción event seleccionada para garantizar que el evento se maneje correctamente dentro de tu función handler.

El ejemplo a continuación demuestra el código CDK para invocar tu función lambda en un horario usando EventBridge:

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

Este generador crea un módulo Terraform para desplegar tu función en la carpeta common/terraform. Instancia el módulo compartido core/asset-bucket una vez por despliegue y pasa su bucket_name en la entrada 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
  ]
}

Esto configura:

1. Función AWS Lambda
2. Grupo de logs de CloudWatch
3. Configuración de rastreo X-Ray
4. Espacio de nombres de métricas de CloudWatch

Esta función puede entonces ser usada como objetivo para cualquier fuente de eventos de lambda. El ejemplo a continuación demuestra el código Terraform para invocar tu función lambda en un horario usando EventBridge:

# 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
}

Coincidencia de Fuente de Eventos

Asegúrate de que la fuente de eventos coincida con la opción event seleccionada para garantizar que el evento se maneje correctamente dentro de tu función handler.