Funciones Lambda de Python

El generador de Funciones Lambda en Python permite agregar una función lambda a un proyecto existente de Python.

Este generador crea un nuevo manejador lambda en Python con configuración de infraestructura AWS CDK. El backend generado utiliza AWS Lambda para despliegues serverless, con opción de validación de tipos usando el Parser de AWS Lambda Powertools. Configura AWS Lambda Powertools para observabilidad, incluyendo registro de logs, trazado con AWS X-Ray y métricas en CloudWatch.

Uso

Generar una función Lambda

Puedes generar una nueva función Lambda de dos formas:

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

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

Opciones

Parámetro	Tipo	Predeterminado	Descripción
project Requerido	string	-	The project to add the lambda function to
functionName Requerido	string	-	The name of the function to add
functionPath	string	-	The path within the project source directory to add the function to
eventSource Requerido	string	Any	Optional event source model to use for the lambda function

Resultado del generador

El generador agregará los siguientes archivos a tu proyecto:

• Directory<module-name>

  • <lambda-function>.py Implementación de la función

El generador también crea constructos CDK para desplegar tu función, ubicados en el directorio packages/common/constructs.

Si se provee la opción functionPath, el generador agregará los archivos en la ruta especificada:

• Directory<module-name>

  • Directory<custom-path>

    • <function-name>.py Implementación de la función

Implementando tu función

La implementación principal está en <function-name>.py. Aquí 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 automáticamente:

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

Observabilidad con AWS Lambda Powertools

Registro de logs

El generador configura registro estructurado usando AWS Lambda Powertools.

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

Consejo

Se recomienda establecer un ID de correlación para todas las solicitudes únicas para facilitar depuración y monitoreo. Consulta la documentación de aws powertools logger para mejores prácticas.

El logger incluye automáticamente:

• Solicitudes de eventos
• Información del contexto Lambda
• Indicadores de cold start

Trazado

El trazado con AWS X-Ray se configura automáticamente. Puedes agregar subsegmentos personalizados:

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    # Crea un nuevo subsegmento
    with tracer.provider.in_subsegment("function-subsegment"):
        # Tu lógica aquí
        return ....

Métricas

Las métricas de CloudWatch se recolectan automáticamente. Puedes agregar métricas personalizadas:

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

Métricas por defecto incluyen:

• Conteo de invocaciones
• Conteos de éxito/fallo
• Métricas de cold start

Validación de tipos

Si elegiste un eventSource al generar tu función, esta se instrumenta con @event_parser de AWS Lambda Powertools. Ejemplo:

@event_parser(model=EventBridgeModel)
def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    event.detail_type # <- tipado seguro con autocompletado en IDE

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

Consejo

Si tienes datos personalizados anidados en un evento, como un stream de DynamoDB o evento de EventBridge, puedes usar Envelopes para garantizar tipado seguro.

Si no deseas tipar tu evento, selecciona Any como eventSource.

Desplegando tu función

El generador crea un constructo CDK para despliegue en common/constructs. Puedes usarlo en una aplicación CDK:

import { MyProjectMyFunction } from ':my-scope/common-constructs';

export class ExampleStack extends Stack {
  constructor(scope: Construct, id: string) {
    // Agrega la función al stack
    const fn = new MyProjectMyFunction(this, 'MyFunction');
  }
}

Esto configura:

1. Función AWS Lambda
2. Grupo de logs en CloudWatch
3. Configuración de trazado X-Ray
4. Namespace para métricas CloudWatch

La función puede usarse como objetivo para cualquier origen de eventos lambda:

Nota

Asegúrate que el origen de eventos coincida con la opción eventSource seleccionada para un manejo adecuado en tu función.

Este ejemplo muestra código CDK para invocar tu función en un horario usando Event Bridge:

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) {
    // Agrega la función al stack
    const fn = new MyProjectMyFunction(this, 'MyFunction');

    // Configura una regla de EventBridge
    const eventRule = new Rule(this, 'MyFunctionScheduleRule', {
      schedule: Schedule.cron({ minute: '15' }),
      targets: [new LambdaFunction(fn)],
    });
  }
}