Funções Lambda em Python

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

Este gerador cria um novo handler de lambda em Python com configuração de infraestrutura AWS CDK ou Terraform. O backend gerado utiliza AWS Lambda para implantação serverless, com opção de type-safety usando o Parser do AWS Lambda Powertools. Configura o AWS Lambda Powertools para observabilidade, incluindo logging, rastreamento com AWS X-Ray e métricas no Cloudwatch.

Utilização

Gerar uma Função Lambda

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

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

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

Opções

Parâmetro	Tipo	Padrão	Descrição
project Obrigatório	string	-	The project to add the lambda function to
functionName Obrigatório	string	-	The name of the function to add
functionPath	string	-	The path within the project source directory to add the function to
eventSource	string	Any	Optional event source model to use for the lambda function
iacProvider	string	Inherit	The preferred IaC provider. By default this is inherited from your initial selection.

Saída do Gerador

O gerador adicionará os seguintes arquivos ao seu projeto:

• Directory<module-name>

  • <lambda-function>.py Implementação da função

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

• Directory<module-name>

  • Directory<custom-path>

    • <function-name>.py Implementação da função

Infraestrutura

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

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

CDK

• Directorypackages/common/constructs

  • Directorysrc

    • Directoryapp/ Constructs para infraestrutura específica de um projeto/gerador

      • …
    • Directorycore/ Constructs genéricos reutilizados pelos constructs em app

      • …
    • index.ts Ponto de entrada exportando os constructs de app
  • project.json Metas de build e configuração do projeto

Nota

Este projeto é gerado usando o gerador ts#project e portanto configura as mesmas metas de build.

Terraform

• Directorypackages/common/terraform

  • Directorysrc

    • Directoryapp/ Módulos Terraform para infraestrutura específica de um projeto/gerador

      • …
    • Directorycore/ Módulos genéricos reutilizados pelos módulos em app

      • …
  • project.json Metas de build e configuração do projeto

Nota

Este projeto é gerado usando o gerador terraform#project e portanto configura as mesmas metas de build.

O gerador cria infraestrutura como código para implantação da função baseado no iacProvider selecionado:

CDK

O gerador cria constructs CDK para implantação da função, localizados no diretório packages/common/constructs.

Terraform

O gerador cria um módulo Terraform para implantação da função, localizado no diretório packages/common/terraform/src/app/lambda-functions/<function-name>.

Implementando sua Função

A implementação principal está em <function-name>.py. Veja 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: Implementar
        metrics.add_metric(name="SuccessCount", unit=MetricUnit.Count, value=1)
        # TODO: Implementar resposta de sucesso se necessário
    except Exception as e:
        logger.exception(e)
        metrics.add_metric(name="ErrorCount", unit=MetricUnit.Count, value=1)
        # TODO: Implementar resposta de erro se necessário

O gerador configura automaticamente vários recursos:

1. Integração com AWS Lambda Powertools para observabilidade
2. Coleta de métricas
3. Type-safety 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()})

Dica

Recomenda-se definir um correlation id para requisições únicas para facilitar depuração e monitoramento. Consulte a documentação do aws powertools logger para melhores práticas.

O logger inclui automaticamente:

• Requisições de evento
• Informações de contexto do Lambda
• Indicadores de cold start

Tracing

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

def lambda_handler(event: EventBridgeModel, context: LambdaContext):
    # Cria um novo subsegmento
    with tracer.provider.in_subsegment("function-subsegment"):
        # Sua lógica aqui
        return ....

Métricas

Métricas no CloudWatch são coletadas automaticamente para cada requisiçã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 ...

Métricas padrão incluem:

• Contagem de invocações
• Contagem de sucessos/falhas
• Métricas de cold start

Type Safety

Se você escolheu um eventSource ao gerar sua função, ela será instrumentada com @event_parser do AWS Lambda Powertools. Exemplo:

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

Isso permite definir modelos de dados usando Pydantic, de forma similar ao trabalho com Fast API.

Dica

Se você possui dados personalizados aninhados em eventos, como streams DynamoDB ou eventos EventBridge, pode beneficiar-se do uso de Envelopes para garantir type-safety.

Se não deseja tipar seu evento, basta selecionar Any como eventSource.

Bundling

O gerador configura automaticamente o empacotamento Python para pacotes de implantação Lambda usando uv:

pnpm

pnpm nx run <project-name>:bundle

yarn

yarn nx run <project-name>:bundle

npm

npx nx run <project-name>:bundle

bun

bunx nx run <project-name>:bundle

Nota

Isso empacota todo o projeto como pacote de implantação. Para minimizar o tamanho, você pode dividir funções lambda em múltiplos projetos Python usando o gerador py#project.

Este processo utiliza:

1. uv export para exportar dependências para um arquivo requirements.txt
2. uv pip install para instalar dependências na plataforma alvo (x86_64-manylinux2014) para implantação Lambda

Nota

Se alterar a arquitetura Lambda para ARM, será necessário editar o target do bundle para usar aarch64-manylinux2014.

Implantando sua Função

Este gerador cria infraestrutura como código CDK ou Terraform com base no iacProvider 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) {
    // Adiciona a função à sua stack
    const fn = new MyProjectMyFunction(this, 'MyFunction');
  }
}

Isso configura:

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

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

Nota

Certifique-se de que a fonte de evento corresponda à opção eventSource selecionada para garantir que o evento seja processado corretamente dentro de sua função handler.

O exemplo abaixo demonstra o código CDK para invocar sua função lambda em um agendamento usando o 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) {
    // Adiciona a função à sua stack
    const fn = new MyProjectMyFunction(this, 'MyFunction');

    // Adiciona a função a uma regra agendada do EventBridge
    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. Você pode usar isso em uma configuração Terraform:

module "my_project_my_function" {
  source = "../../common/terraform/src/app/lambda-functions/my-project-my-function"

  # Variáveis de ambiente para a função Lambda
  env = {
    SOME_VARIABLE = "some value"
  }

  # Políticas IAM adicionais se necessário
  additional_iam_policy_statements = [
    # Adicione quaisquer permissões adicionais necessárias
  ]
}

Isso configura:

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

Esta função pode 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 agendamento usando o EventBridge:

# Regra do EventBridge para execução agendada
resource "aws_cloudwatch_event_rule" "my_function_schedule" {
  name                = "my-function-schedule"
  description         = "Aciona minha função a cada 15 minutos"
  schedule_expression = "cron(15 * * * ? *)"
}

# Target do EventBridge
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
}

# Permissão para o EventBridge invocar a função Lambda
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
}

Nota

Certifique-se de que a fonte de evento corresponda à opção eventSource selecionada para garantir que o evento seja processado corretamente dentro de sua função handler.