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 :

Exécuter ce générateur@aws/nx-plugin:py#lambda-function

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

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

Composez votre commande7

projectRequis

nameRequis

event

AnyAlbModelAPIGatewayProxyEventModelApiGatewayAuthorizerTokenApiGatewayAuthorizerRequestAPIGatewayProxyEventV2ModelApiGatewayAuthorizerRequestV2APIGatewayWebSocketMessageEventModelAPIGatewayWebSocketConnectEventModelAPIGatewayWebSocketDisconnectEventModelBedrockAgentEventModelCloudFormationCustomResourceCreateModelCloudFormationCustomResourceUpdateModelCloudFormationCustomResourceDeleteModelCloudWatchLogsModelDynamoDBStreamModelEventBridgeModelIoTCoreThingEventIoTCoreThingTypeEventIoTCoreThingTypeAssociationEventIoTCoreThingGroupEventIoTCoreAddOrRemoveFromThingGroupEventIoTCoreAddOrDeleteFromThingGroupEventKafkaMskEventModelKafkaSelfManagedEventModelKinesisDataStreamModelKinesisFirehoseModelKinesisFirehoseSqsModelLambdaFunctionUrlModelS3BatchOperationModelS3EventNotificationEventBridgeModelS3ModelS3ObjectLambdaEventS3SqsEventNotificationModelSesModelSnsModelSqsModelTransferFamilyAuthorizerVpcLatticeModelVpcLatticeV2ModelTout afficher 40

infra

lambdanone

iac

inheritcdkterraform

functionPath

preferInstallDependencies

true

--dry-runVous pouvez également effectuer une simulation pour voir quels fichiers seraient modifiés

Options

Options du générateur7 options

projectRequisstring

Le projet auquel ajouter la fonction lambda

nameRequisstring

Le nom de la fonction à ajouter

eventenumPar défaut: Any

Modèle de source d'événement optionnel à utiliser pour la fonction lambda

AnyAlbModelAPIGatewayProxyEventModelApiGatewayAuthorizerTokenApiGatewayAuthorizerRequestAPIGatewayProxyEventV2ModelApiGatewayAuthorizerRequestV2APIGatewayWebSocketMessageEventModelAPIGatewayWebSocketConnectEventModelAPIGatewayWebSocketDisconnectEventModelBedrockAgentEventModelCloudFormationCustomResourceCreateModelCloudFormationCustomResourceUpdateModelCloudFormationCustomResourceDeleteModelCloudWatchLogsModelDynamoDBStreamModelEventBridgeModelIoTCoreThingEventIoTCoreThingTypeEventIoTCoreThingTypeAssociationEventIoTCoreThingGroupEventIoTCoreAddOrRemoveFromThingGroupEventIoTCoreAddOrDeleteFromThingGroupEventKafkaMskEventModelKafkaSelfManagedEventModelKinesisDataStreamModelKinesisFirehoseModelKinesisFirehoseSqsModelLambdaFunctionUrlModelS3BatchOperationModelS3EventNotificationEventBridgeModelS3ModelS3ObjectLambdaEventS3SqsEventNotificationModelSesModelSnsModelSqsModelTransferFamilyAuthorizerVpcLatticeModelVpcLatticeV2ModelTout afficher 40

infraenumPar défaut: lambda

Le type d'infrastructure pour déployer votre fonction Lambda.

lambdanone

iacenumPar défaut: inherit

Le fournisseur IaC préféré. Par défaut, celui-ci est hérité de votre sélection initiale.

inheritcdkterraform

functionPathstring

Le chemin dans le répertoire source du projet où ajouter la fonction

preferInstallDependenciesbooleanPar défaut: 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, avec ses logs, métriques et traces envoyés vers CloudWatch et X-Ray :

Loading the diagram…

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
@logger.inject_lambda_context
@metrics.log_metrics(capture_cold_start_metric=True)
@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 gestionnaire est décoré avec @logger.inject_lambda_context, donc chaque enregistrement de journal qu’il émet contient également :

• function_name, function_memory_size, function_arn et function_request_id du contexte Lambda
• cold_start, indiquant si l’invocation a initialisé un nouvel environnement d’exécution

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 :

• InvocationCount, ajouté à chaque invocation
• SuccessCount et ErrorCount, ajoutés par le try/except du gestionnaire
• ColdStart, émis par @metrics.log_metrics(capture_cold_start_metric=True) lorsque l’invocation initialise un nouvel environnement d’exécution

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.