Saltearse al contenido

Funciones Lambda de Python

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

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

Uso

Generar una función Lambda

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

  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
    • Haga clic en Generate

    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

    Salida del generador

    El generador añadirá los siguientes archivos a tu proyecto:

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

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

    Si se provee la opción functionPath, el generador añadirá 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

    Logging

    El generador configura logging estructurado usando AWS Lambda Powertools.

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

    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 añadir 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 añadir 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
    • Conteo de éxitos/fallos
    • Métricas de cold start

    Validación de tipos

    Si seleccionaste un eventSource al generar tu función, esta se instrumenta con @event_parser de AWS Lambda Powertools. Por 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.

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

    Implementando tu función

    El generador crea un construct CDK en common/constructs para implementar tu función. Puedes usarlo en una aplicación CDK:

    import { MyProjectMyFunction } from ':my-scope/common-constructs';
    export class ExampleStack extends Stack {
    constructor(scope: Construct, id: string) {
    // Añade la función a tu 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 de métricas en CloudWatch

    La función puede usarse como destino para cualquier origen de evento de Lambda:

    Este ejemplo muestra código CDK para invocar tu función en un schedule 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) {
    // Añade la función al stack
    const fn = new MyProjectMyFunction(this, 'MyFunction');
    // Añade la función a una regla de EventBridge
    const eventRule = new Rule(this, 'MyFunctionScheduleRule', {
    schedule: Schedule.cron({ minute: '15' }),
    targets: [new LambdaFunction(fn)],
    });
    }
    }