Funções Lambda em Python

O gerador de Função Lambda em Python fornece a capacidade de adicionar uma função lambda a um projeto python existente.

Este gerador cria um novo manipulador lambda python com configuração de infraestrutura AWS CDK ou Terraform. O backend gerado usa AWS Lambda para implantação serverless, com segurança de tipo opcional usando o Parser do AWS Lambda Powertools. Ele configura o AWS Lambda Powertools para observabilidade, incluindo logging, rastreamento AWS X-Ray e Cloudwatch Metrics.

Uso

Gerar uma Função Lambda

Você pode gerar uma nova Função Lambda de duas maneiras:

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

Você também pode realizar uma execução simulada para ver quais arquivos seriam alterados

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. Instale o Nx Console VSCode Plugin se ainda não o fez
2. Abra o console Nx no VSCode
3. Clique em Generate (UI) na seção "Common Nx Commands"
4. Procure por @aws/nx-plugin - py#lambda-function
5. Preencha os parâmetros obrigatórios
6. Clique em Generate

Opções

Parâmetro	Tipo	Padrão	Descrição
project Obrigatório	string	-	O projeto ao qual adicionar a função lambda
name Obrigatório	string	-	O nome da função a adicionar
functionPath	string	-	O caminho dentro do diretório de origem do projeto para adicionar a função
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	Modelo de origem de evento opcional para usar na função lambda
infra	lambda | none	lambda	O tipo de infraestrutura para implantar sua função Lambda.
iac	inherit | cdk | terraform	inherit	O provedor IaC preferido. Por padrão, isso é herdado da sua seleção inicial.
preferInstallDependencies	boolean	true	Se deve preferir instalar dependências após a execução do gerador. Defina como false para adiar a instalação ao executar múltiplos geradores em lote (uma instalação ainda é executada se necessário para que geradores subsequentes possam calcular o grafo de projetos Nx); instale uma vez no final.

Saída do Gerador

O gerador adicionará os seguintes arquivos ao seu projeto:

• Directory<module-name>

  • <lambda-function>.py Function implementation

Se a opção functionPath for fornecida, o gerador adicionará os arquivos necessários ao caminho especificado:

• Directory<module-name>

  • Directory<custom-path>

    • <function-name>.py Function implementation

Infraestrutura

Como este gerador fornece infraestrutura como código baseada no seu iac escolhido, ele criará um projeto em packages/common que inclui os constructs CDK relevantes ou módulos Terraform.

O projeto comum de infraestrutura como código é estruturado da seguinte forma:

CDK

• Directorypackages/common/constructs

  • Directorysrc

    • Directoryapp/ Constructs for infrastructure specific to a project/generator

      • …
    • Directorycore/ Generic constructs which are reused by constructs in app

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

Projeto Gerado

Este projeto é gerado usando o gerador ts#project e, portanto, configura os mesmos alvos de build.

Terraform

• Directorypackages/common/terraform

  • Directorysrc

    • Directoryapp/ Terraform modules for infrastructure specific to a project/generator

      • …
    • Directorycore/ Generic modules which are reused by modules in app

      • …
  • project.json Project build targets and configuration

Projeto Gerado

Este projeto é gerado usando o gerador terraform#project e, portanto, configura os mesmos alvos de build.

O gerador cria infraestrutura como código para implantar sua função com base no iac selecionado:

CDK

O gerador cria constructs CDK que podem ser usados para implantar sua função, que residem no diretório packages/common/constructs.

Terraform

O gerador cria um módulo Terraform que pode ser usado para implantar sua função, que reside no diretório packages/common/terraform/src/app/lambda-functions/<function-name>.

Arquitetura

A função implantada possui a seguinte arquitetura:

O gerador fornece a função em si; você conecta a origem do evento em sua stack para invocá-la (por exemplo, uma integração com API Gateway, uma regra do EventBridge, uma notificação do S3 ou um mapeamento de origem de eventos do SQS).

Implementando sua Função

A implementação principal da função está em <function-name>.py. Aqui está um exemplo:

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

O gerador configura vários recursos automaticamente:

1. Integração com AWS Lambda Powertools para observabilidade
2. Coleta de métricas
3. Segurança de tipo usando @event_parser

Observabilidade com AWS Lambda Powertools

Logging

O gerador configura logging estruturado usando AWS Lambda Powertools.

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

ID de Correlação

É recomendado definir o id de correlação para todas as solicitações únicas para facilitar a depuração e monitoramento. Consulte a documentação do aws powertools logger para as melhores práticas de id de correlação.

O logger inclui automaticamente:

• Solicitações de eventos
• Informações de contexto do Lambda
• Indicadores de cold start

Rastreamento

O rastreamento AWS X-Ray é configurado automaticamente. Você pode adicionar subsegmentos personalizados aos seus rastreamentos:

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    # Creates a new subsegment
    with tracer.provider.in_subsegment("function-subsegment"):
        # Your logic here
        return ....

Métricas

As métricas do CloudWatch são coletadas automaticamente para cada solicitação. Você pode adicionar métricas personalizadas:

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

As métricas padrão incluem:

• Contagens de invocação
• Contagens de sucesso/falha
• Métricas de cold start

Segurança de Tipo

event ≠ Any

Se você escolheu um event ao gerar sua função lambda, sua função é instrumentada com @event_parser do AWS Lambda Powertools. Por exemplo:

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

Isso permite que você defina modelos de dados usando Pydantic, de maneira similar a trabalhar com Fast API.

Dados de Evento Personalizados

Se você tiver dados personalizados aninhados dentro de um evento, por exemplo um stream do DynamoDB ou evento do EventBridge, você pode se beneficiar do uso de Envelopes para fornecer segurança de tipo para esses dados personalizados.

event = Any

Se você selecionou Any para seu event, o decorador @event_parser não é aplicado e o manipulador recebe um dict de evento sem tipo. Regenere com um event específico se você quiser validação Pydantic e acesso a campos tipados.

Empacotamento

O gerador configura automaticamente o empacotamento Python para pacotes de implantação do Lambda usando 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>

Comportamento do Empacotamento

Isso empacota todo o projeto como o pacote de implantação. Para minimizar o tamanho da implantação, você pode desejar dividir várias funções lambda em vários projetos Python gerados com o gerador py#project.

Este processo usa:

1. uv export para exportar suas dependências Python para um arquivo requirements.txt
2. uv pip install para instalar dependências para a plataforma de destino (x86_64-manylinux_2_28) para implantação no Lambda

Mudanças de Arquitetura

Se você alterar sua arquitetura lambda para ARM, você precisará editar seu destino de empacotamento para usar o destino aarch64-manylinux_2_28.

Implantando sua Função

Este gerador cria infraestrutura como código CDK ou Terraform com base no seu iac selecionado. Você pode usar isso para implantar sua função.

CDK

Este gerador cria um construct CDK para implantar sua função na pasta common/constructs. Você pode usar isso em uma aplicação 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');
  }
}

Isso configura:

1. Função AWS Lambda
2. Grupo de logs CloudWatch
3. Configuração de rastreamento X-Ray
4. Namespace de métricas CloudWatch

Esta função pode então ser usada como destino para qualquer fonte de evento lambda:

Correspondência de Fonte de Evento

Certifique-se de que a fonte de evento corresponda à opção event selecionada para garantir que o evento seja tratado adequadamente dentro da sua função handler.

O exemplo abaixo demonstra o código CDK para invocar sua função lambda em um cronograma usando 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

Este gerador cria um módulo Terraform para implantar sua função na pasta common/terraform. Instancie o módulo compartilhado core/asset-bucket uma vez por implantação e passe seu bucket_name para a entrada 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
  ]
}

Isso configura:

1. Função AWS Lambda
2. Grupo de logs CloudWatch
3. Configuração de rastreamento X-Ray
4. Namespace de métricas CloudWatch

Esta função pode então ser usada como destino para qualquer fonte de evento lambda. O exemplo abaixo demonstra o código Terraform para invocar sua função lambda em um cronograma usando 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
}

Correspondência de Fonte de Evento

Certifique-se de que a fonte de evento corresponda à opção event selecionada para garantir que o evento seja tratado adequadamente dentro da sua função handler.