Fonctions Lambda Python

Le générateur de fonctions Lambda Python offre la possibilité d’ajouter une fonction lambda à un projet python existant.

Ce générateur crée un nouveau gestionnaire lambda python avec une configuration d’infrastructure AWS CDK ou Terraform. Le backend généré utilise AWS Lambda pour un déploiement serverless, avec une sécurité de type optionnelle utilisant le Parser from AWS Lambda Powertools. Il configure AWS Lambda Powertools pour l’observabilité, incluant la journalisation, le traçage 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 :

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

Vous pouvez également effectuer une simulation pour voir quels fichiers seraient modifiés

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. 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
6. Cliquez sur Generate

Options

Paramètre	Type	Par défaut	Description
project Requis	string	-	Le projet auquel ajouter la fonction lambda
name Requis	string	-	Le nom de la fonction à ajouter
functionPath	string	-	Le chemin dans le répertoire source du projet où ajouter la fonction
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	Modèle de source d'événement optionnel à utiliser pour la fonction lambda
infra	lambda | none	lambda	Le type d'infrastructure pour déployer votre fonction Lambda.
iac	inherit | cdk | terraform	inherit	Le fournisseur IaC préféré. Par défaut, celui-ci est hérité de votre sélection initiale.
preferInstallDependencies	boolean	true	Indique s'il faut privilégier l'installation des dépendances après l'exécution du générateur. Définir sur false pour différer l'installation lors de l'exécution de plusieurs générateurs en lot (une installation s'exécute quand même si nécessaire pour que les générateurs suivants puissent calculer le graphe de projet Nx) ; installer une seule fois à la fin.

Sortie du générateur

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

• Répertoire<module-name>

  • <lambda-function>.py Function implementation

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

• Répertoire<module-name>

  • Répertoire<custom-path>

    • <function-name>.py Function implementation

Infrastructure

Étant donné que ce générateur fournit de l’infrastructure en tant que code basée sur votre iac choisi, il créera un projet dans packages/common qui inclut les constructs CDK ou modules Terraform pertinents.

Le projet d’infrastructure en tant que code commun est structuré comme suit :

CDK

• Répertoirepackages/common/constructs

  • Répertoiresrc

    • Répertoireapp/ Constructs pour l’infrastructure spécifique à un projet/générateur

      • …
    • Répertoirecore/ Constructs génériques qui sont réutilisés par les constructs dans app

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

Projet généré

Ce projet est généré en utilisant le générateur ts#project et configure donc les mêmes cibles de build.

Terraform

• Répertoirepackages/common/terraform

  • Répertoiresrc

    • Répertoireapp/ Terraform modules for infrastructure specific to a project/generator

      • …
    • Répertoirecore/ Generic modules which are reused by modules in app

      • …
  • project.json Project build targets and configuration

Projet généré

Ce projet est généré en utilisant le générateur terraform#project et configure donc les mêmes cibles de build.

Le générateur crée une infrastructure en tant que code pour déployer votre fonction en fonction de votre iac sélectionné :

CDK

Le générateur crée des constructs CDK qui peuvent être utilisés pour déployer votre fonction, qui résident dans le répertoire packages/common/constructs.

Terraform

Le générateur crée un module Terraform qui peut être utilisé pour déployer votre fonction, qui réside dans le répertoire packages/common/terraform/src/app/lambda-functions/<function-name>.

Architecture

La fonction déployée a l’architecture suivante :

Le générateur fournit la fonction elle-même ; vous connectez la source d’événements dans votre stack pour l’invoquer (par exemple, une intégration API Gateway, une règle EventBridge, une notification S3 ou un mappage de source d’événements SQS).

Implémenter votre fonction

L’implémentation principale de la fonction 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: 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

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. Sécurité de type utilisant @event_parser

Observabilité avec AWS Lambda Powertools

Journalisation

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

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

ID de corrélation

Il est recommandé de définir l’ID de corrélation pour toutes les requêtes uniques afin de faciliter le débogage et la surveillance. Veuillez vous référer à la documentation aws powertools logger pour les meilleures pratiques concernant l’ID de corrélation.

Le logger inclut automatiquement :

• Les requêtes d’événements
• Les informations de contexte Lambda
• Les indicateurs de démarrage à froid

Traçage

Le traçage AWS X-Ray est configuré automatiquement. Vous pouvez ajouter des sous-segments personnalisés à vos traces :

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    # Creates a new subsegment
    with tracer.provider.in_subsegment("function-subsegment"):
        # Your logic here
        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 :

• Compteurs d’invocations
• Compteurs de succès/échec
• Métriques de démarrage à froid

Sécurité de type

event ≠ Any

Si vous avez choisi un event lors de la génération de votre fonction lambda, votre fonction est instrumentée avec @event_parser from AWS Lambda Powertools. Par exemple :

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

Cela vous permet de définir des modèles de données en utilisant Pydantic, de manière similaire au travail avec Fast API.

Données d’événement personnalisées

Si vous avez des données personnalisées imbriquées dans un événement, par exemple un flux DynamoDB ou un événement EventBridge, vous pourriez bénéficier de l’utilisation d’Envelopes pour fournir une sécurité de type pour ces données personnalisées.

event = Any

Si vous avez sélectionné Any pour votre event, le décorateur @event_parser n’est pas appliqué et le gestionnaire reçoit un dictionnaire d’événement non typé. Régénérez avec un event spécifique si vous souhaitez une validation Pydantic et un accès aux champs typés.

Bundling

Le générateur configure automatiquement le bundling Python pour les packages de déploiement Lambda en utilisant 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>

Comportement du bundle

Cela regroupe l’ensemble du projet en tant que package de déploiement. Pour minimiser la taille du déploiement, vous pouvez souhaiter diviser plusieurs fonctions lambda sur plusieurs projets Python générés avec le générateur py#project.

Ce processus utilise :

1. uv export pour exporter vos dépendances Python vers un fichier requirements.txt
2. uv pip install pour installer les dépendances pour la plateforme cible (x86_64-manylinux_2_28) pour le déploiement Lambda

Changements d’architecture

Si vous changez votre architecture lambda vers ARM, vous devrez modifier votre cible de bundle pour utiliser la cible aarch64-manylinux_2_28.

Déployer votre fonction

Ce générateur crée du code d’infrastructure CDK ou Terraform en fonction de votre iac sélectionné. Vous pouvez l’utiliser pour déployer votre fonction.

CDK

Ce générateur crée une construction CDK pour déployer votre fonction dans le dossier common/constructs. Vous pouvez l’utiliser dans une application 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');
  }
}

Cela configure :

1. Fonction AWS Lambda
2. Groupe de journaux CloudWatch
3. Configuration du traçage X-Ray
4. Espace de noms des métriques CloudWatch

Cette fonction peut ensuite être utilisée comme cible pour n’importe quelle source d’événements lambda :

Correspondance de la source d’événements

Assurez-vous que la source d’événements correspond à l’option event sélectionnée pour garantir que l’événement est correctement géré dans votre fonction de gestionnaire.

L’exemple ci-dessous démontre le code CDK pour invoquer votre fonction lambda selon un calendrier en utilisant 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

Ce générateur crée un module Terraform pour déployer votre fonction dans le dossier common/terraform. Instanciez le module partagé core/asset-bucket une fois par déploiement et transmettez son bucket_name dans l’entrée 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
  ]
}

Cela configure :

1. Fonction AWS Lambda
2. Groupe de journaux CloudWatch
3. Configuration du traçage X-Ray
4. Espace de noms des métriques CloudWatch

Cette fonction peut ensuite être utilisée comme cible pour n’importe quelle source d’événements lambda. L’exemple ci-dessous démontre le code Terraform pour invoquer votre fonction lambda selon un calendrier en utilisant 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
}

Correspondance de la source d’événements

Assurez-vous que la source d’événements correspond à l’option event sélectionnée pour garantir que l’événement est correctement géré dans votre fonction de gestionnaire.