Aller au contenu
@aws/nx-plugin
Blog

Fonctions Lambda Python

Le générateur de fonction Lambda Python permet d’ajouter une fonction lambda à un projet Python existant.

Ce générateur crée un nouveau gestionnaire de fonction lambda Python avec une configuration d’infrastructure AWS CDK. Le backend généré utilise AWS Lambda pour un déploiement serverless, avec une validation de types optionnelle via le Parser d’AWS Lambda Powertools. Il configure AWS Lambda Powertools pour l’observabilité, incluant la journalisation, le tracing AWS X-Ray et les métriques CloudWatch.

Utilisation

Générer une fonction Lambda

Vous pouvez générer une nouvelle fonction Lambda de deux manières :

  1. Installez le Nx Console VSCode Plugin si ce n'est pas déjà fait
  2. Ouvrez la console Nx dans VSCode
  3. Cliquez sur Generate (UI) dans la section "Common Nx Commands"
  4. Recherchez @aws/nx-plugin - py#lambda-function
  5. Remplissez les paramètres requis
    • Cliquez sur Generate

    Options

    Paramètre Type Par défaut Description
    project Requis string - The project to add the lambda function to
    functionName Requis string - The name of the function to add
    functionPath string - The path within the project source directory to add the function to
    eventSource Requis string Any Optional event source model to use for the lambda function

    Résultat du générateur

    Le générateur ajoutera les fichiers suivants à votre projet :

    • Répertoire<module-name>
      • <lambda-function>.py Implémentation de la fonction

    Le générateur crée également des constructs CDK pour déployer votre fonction, situés dans le répertoire packages/common/constructs.

    Si l’option functionPath est spécifiée, le générateur ajoutera les fichiers nécessaires au chemin indiqué :

    • Répertoire<module-name>
      • Répertoire<custom-path>
        • <function-name>.py Implémentation de la fonction

    Implémentation de votre fonction

    L’implémentation principale se trouve dans <function-name>.py. Voici un exemple :

    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: Implémenter
            metrics.add_metric(name="SuccessCount", unit=MetricUnit.Count, value=1)
            # TODO: Implémenter la réponse en cas de succès si nécessaire
        except Exception as e:
            logger.exception(e)
            metrics.add_metric(name="ErrorCount", unit=MetricUnit.Count, value=1)
            # TODO: Implémenter la réponse en cas d'erreur si nécessaire

    Le générateur configure automatiquement plusieurs fonctionnalités :

    1. Intégration d’AWS Lambda Powertools pour l’observabilité
    2. Collecte de métriques
    3. Validation de types avec @event_parser

    Observabilité avec AWS Lambda Powertools

    Journalisation

    Le générateur configure la journalisation structurée avec AWS Lambda Powertools.

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

    Il est recommandé de définir un identifiant de corrélation pour chaque requête unique afin de faciliter le débogage et le monitoring. Consultez la documentation aws powertools logger pour les bonnes pratiques.

    Le logger inclut automatiquement :

    • Les requêtes d’événements
    • Les informations de contexte Lambda
    • Les indicateurs de cold start

    Tracing

    Le tracing AWS X-Ray est configuré automatiquement. Vous pouvez ajouter des sous-segments personnalisés :

    def lambda_handler(event: EventBridgeModel, context: LambdaContext):
        # Crée un nouveau sous-segment
        with tracer.provider.in_subsegment("function-subsegment"):
            # Votre logique ici
            return ....

    Métriques

    Les métriques CloudWatch sont collectées automatiquement pour chaque requête. Vous pouvez ajouter des métriques personnalisées :

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

    Les métriques par défaut incluent :

    • Nombre d’invocations
    • Taux de succès/échec
    • Métriques de cold start

    Validation de types

    Si vous avez choisi une eventSource lors de la génération, votre fonction utilise @event_parser d’AWS Lambda Powertools. Exemple :

    @event_parser(model=EventBridgeModel)
    def lambda_handler(event: EventBridgeModel, context: LambdaContext):
        event.detail_type # <- typage fort avec autocomplétion IDE

    Ceci permet de définir des modèles de données avec Pydantic, similaire à Fast API.

    Pour des données personnalisées dans un événement (ex: flux DynamoDB ou événement EventBridge), utilisez Envelopes pour assurer le typage.

    Si vous ne souhaitez pas typer votre événement, sélectionnez Any comme eventSource.

    Déploiement de votre fonction

    Le générateur crée un construct CDK dans common/constructs pour déployer votre fonction. Utilisez-le dans une application CDK :

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

    export class ExampleStack extends Stack {
      constructor(scope: Construct, id: string) {
        // Ajoute la fonction à votre stack
        const fn = new MyProjectMyFunction(this, 'MyFunction');
      }
    }

    Ceci configure :

    1. La fonction AWS Lambda
    2. Le groupe de logs CloudWatch
    3. La configuration du tracing X-Ray
    4. L’espace de noms des métriques CloudWatch

    La fonction peut ensuite être utilisée comme cible pour n’importe quelle source d’événement lambda.

    Assurez-vous que la source d’événement correspond à l’option eventSource sélectionnée pour un traitement correct.

    L’exemple suivant montre comment invoquer votre fonction selon un planning avec 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) {
        // Ajoute la fonction à votre stack
        const fn = new MyProjectMyFunction(this, 'MyFunction');
    

        // Configure une règle EventBridge
        const eventRule = new Rule(this, 'MyFunctionScheduleRule', {
          schedule: Schedule.cron({ minute: '15' }),
          targets: [new LambdaFunction(fn)],
        });
      }
    }