1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# Estructura básica de una función Lambda
import json

def lambda_handler(event, context):
    """
    Función Lambda básica
    
    Args:
        event: Datos del evento que activa la función
        context: Información del runtime y la función
    
    Returns:
        dict: Respuesta de la función
    """
    
    # Procesar el evento
    name = event.get('name', 'World')
    
    # Lógica de negocio
    message = f"Hello, {name}!"
    
    # Retornar respuesta
    return {
        'statusCode': 200,
        'body': json.dumps({
            'message': message
        })
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

# template.yaml (SAM)
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Resources:
  UserAPI:
    Type: AWS::Serverless::Api
    Properties:
      StageName: prod
      Cors:
        AllowMethods: "'GET,POST,PUT,DELETE'"
        AllowHeaders: "'Content-Type,X-Amz-Date,Authorization'"
        AllowOrigin: "'*'"

  GetUsersFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: users.get_users
      Runtime: python3.11
      Events:
        GetUsers:
          Type: Api
          Properties:
            RestApiId: !Ref UserAPI
            Path: /users
            Method: get

  CreateUserFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: users.create_user
      Runtime: python3.11
      Events:
        CreateUser:
          Type: Api
          Properties:
            RestApiId: !Ref UserAPI
            Path: /users
            Method: post

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

# Función Lambda para procesar mensajes SQS
import json
import boto3
from typing import Dict, Any

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('ProcessedEvents')

def lambda_handler(event: Dict[str, Any], context) -> Dict[str, Any]:
    """
    Procesa mensajes de SQS de forma batch
    """
    
    processed_count = 0
    failed_count = 0
    
    for record in event['Records']:
        try:
            # Parsear mensaje SQS
            message_body = json.loads(record['body'])
            
            # Procesar mensaje
            result = process_message(message_body)
            
            # Guardar en DynamoDB
            table.put_item(Item={
                'id': message_body['id'],
                'processed_at': context.aws_request_id,
                'result': result,
                'status': 'processed'
            })
            
            processed_count += 1
            
        except Exception as e:
            print(f"Error processing message: {str(e)}")
            failed_count += 1
    
    return {
        'statusCode': 200,
        'body': json.dumps({
            'processed': processed_count,
            'failed': failed_count
        })
    }

def process_message(message: Dict[str, Any]) -> Dict[str, Any]:
    """
    Lógica de procesamiento del mensaje
    """
    # Implementar lógica específica
    return {
        'processed': True,
        'timestamp': message.get('timestamp'),
        'data': message.get('data')
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

# Procesamiento de streams de datos
import json
import base64
import boto3
from datetime import datetime

s3 = boto3.client('s3')
BUCKET_NAME = 'data-lake-bucket'

def lambda_handler(event, context):
    """
    Procesa records de Kinesis Data Streams
    """
    
    processed_records = []
    
    for record in event['Records']:
        # Decodificar datos de Kinesis
        payload = base64.b64decode(record['kinesis']['data'])
        data = json.loads(payload)
        
        # Transformar datos
        transformed_data = transform_data(data)
        
        # Agregar timestamp de procesamiento
        transformed_data['processed_at'] = datetime.utcnow().isoformat()
        
        processed_records.append(transformed_data)
    
    # Guardar batch en S3
    if processed_records:
        save_to_s3(processed_records)
    
    return {
        'statusCode': 200,
        'recordsProcessed': len(processed_records)
    }

def transform_data(raw_data):
    """
    Transforma y enriquece los datos
    """
    return {
        'user_id': raw_data.get('userId'),
        'event_type': raw_data.get('eventType'),
        'timestamp': raw_data.get('timestamp'),
        'metadata': {
            'source': raw_data.get('source', 'unknown'),
            'version': '1.0'
        }
    }

def save_to_s3(records):
    """
    Guarda records procesados en S3
    """
    timestamp = datetime.utcnow().strftime('%Y/%m/%d/%H')
    key = f"processed-data/{timestamp}/{context.aws_request_id}.json"
    
    s3.put_object(
        Bucket=BUCKET_NAME,
        Key=key,
        Body=json.dumps(records),
        ContentType='application/json'
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

import boto3
from botocore.exceptions import ClientError
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    """
    Función con manejo robusto de errores
    """
    
    try:
        # Lógica principal
        result = process_business_logic(event)
        
        return {
            'statusCode': 200,
            'body': json.dumps(result)
        }
        
    except ValidationError as e:
        # Error de validación - no reintentar
        logger.error(f"Validation error: {str(e)}")
        return {
            'statusCode': 400,
            'body': json.dumps({'error': 'Invalid input'})
        }
        
    except ClientError as e:
        # Error de AWS - puede ser temporal
        error_code = e.response['Error']['Code']
        
        if error_code in ['ThrottlingException', 'ServiceUnavailable']:
            # Error temporal - Lambda reintentará automáticamente
            logger.warning(f"Temporary AWS error: {error_code}")
            raise e
        else:
            # Error permanente
            logger.error(f"AWS error: {error_code}")
            return {
                'statusCode': 500,
                'body': json.dumps({'error': 'Service error'})
            }
            
    except Exception as e:
        # Error inesperado
        logger.error(f"Unexpected error: {str(e)}")
        raise e  # Lambda reintentará

class ValidationError(Exception):
    """Excepción personalizada para errores de validación"""
    pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

# Inicialización fuera del handler
import boto3
import json
from typing import Dict, Any

# Conexiones reutilizables
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('MyTable')

# Cache de configuración
config_cache = {}

def lambda_handler(event: Dict[str, Any], context) -> Dict[str, Any]:
    """
    Handler optimizado para cold starts
    """
    
    # Usar conexiones pre-inicializadas
    result = table.get_item(Key={'id': event['id']})
    
    # Usar cache cuando sea posible
    config = get_cached_config()
    
    return {
        'statusCode': 200,
        'body': json.dumps({
            'data': result.get('Item'),
            'config': config
        })
    }

def get_cached_config():
    """
    Obtiene configuración con cache
    """
    if not config_cache:
        # Cargar configuración solo una vez
        config_cache.update({
            'api_version': '1.0',
            'timeout': 30,
            'retry_count': 3
        })
    
    return config_cache

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

import boto3
import json
from aws_lambda_powertools import Logger, Tracer, Metrics
from aws_lambda_powertools.metrics import MetricUnit

# Inicializar herramientas de observabilidad
logger = Logger()
tracer = Tracer()
metrics = Metrics()

@tracer.capture_lambda_handler
@logger.inject_lambda_context
@metrics.log_metrics
def lambda_handler(event, context):
    """
    Handler con observabilidad completa
    """
    
    # Log estructurado
    logger.info("Processing request", extra={
        "request_id": context.aws_request_id,
        "event_type": event.get('eventType')
    })
    
    # Métricas personalizadas
    metrics.add_metric(name="RequestReceived", unit=MetricUnit.Count, value=1)
    
    try:
        # Tracing de operaciones
        with tracer.subsegment("business_logic"):
            result = process_request(event)
        
        # Métrica de éxito
        metrics.add_metric(name="RequestSuccess", unit=MetricUnit.Count, value=1)
        
        logger.info("Request processed successfully", extra={
            "result_count": len(result)
        })
        
        return {
            'statusCode': 200,
            'body': json.dumps(result)
        }
        
    except Exception as e:
        # Métrica de error
        metrics.add_metric(name="RequestError", unit=MetricUnit.Count, value=1)
        
        logger.error("Request failed", extra={
            "error": str(e),
            "error_type": type(e).__name__
        })
        
        raise

@tracer.capture_method
def process_request(event):
    """
    Lógica de negocio con tracing
    """
    # Implementar lógica
    return {"processed": True}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# Arquitectura basada en eventos
Resources:
  # EventBridge para routing de eventos
  EventBus:
    Type: AWS::Events::EventBus
    Properties:
      Name: MyApplicationEventBus

  # Regla para eventos de usuario
  UserEventRule:
    Type: AWS::Events::Rule
    Properties:
      EventBusName: !Ref EventBus
      EventPattern:
        source: ["myapp.users"]
        detail-type: ["User Created", "User Updated"]
      Targets:
        - Arn: !GetAtt UserProcessorFunction.Arn
          Id: "UserProcessor"

  # Función para procesar eventos de usuario
  UserProcessorFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: user_processor.handler
      Runtime: python3.11
      Environment:
        Variables:
          EVENT_BUS_NAME: !Ref EventBus

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

# Command Handler
def create_user_command_handler(event, context):
    """
    Maneja comandos de creación de usuario
    """
    
    command = json.loads(event['body'])
    
    # Validar comando
    validate_create_user_command(command)
    
    # Crear usuario
    user_id = str(uuid.uuid4())
    user = {
        'id': user_id,
        'name': command['name'],
        'email': command['email'],
        'created_at': datetime.utcnow().isoformat()
    }
    
    # Guardar en write store
    write_table.put_item(Item=user)
    
    # Publicar evento
    publish_event('user.created', user)
    
    return {
        'statusCode': 201,
        'body': json.dumps({'user_id': user_id})
    }

# Query Handler
def get_user_query_handler(event, context):
    """
    Maneja queries de lectura de usuario
    """
    
    user_id = event['pathParameters']['id']
    
    # Leer de read store optimizado
    response = read_table.get_item(Key={'id': user_id})
    
    if 'Item' not in response:
        return {
            'statusCode': 404,
            'body': json.dumps({'error': 'User not found'})
        }
    
    return {
        'statusCode': 200,
        'body': json.dumps(response['Item'])
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

import boto3
from PIL import Image
import io

s3 = boto3.client('s3')

def lambda_handler(event, context):
    """
    Procesa imágenes subidas a S3
    """
    
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']
        
        # Descargar imagen original
        response = s3.get_object(Bucket=bucket, Key=key)
        image_data = response['Body'].read()
        
        # Procesar imagen
        thumbnails = create_thumbnails(image_data)
        
        # Subir thumbnails
        for size, thumbnail_data in thumbnails.items():
            thumbnail_key = f"thumbnails/{size}/{key}"
            s3.put_object(
                Bucket=bucket,
                Key=thumbnail_key,
                Body=thumbnail_data,
                ContentType='image/jpeg'
            )

def create_thumbnails(image_data):
    """
    Crea thumbnails de diferentes tamaños
    """
    image = Image.open(io.BytesIO(image_data))
    thumbnails = {}
    
    sizes = [(150, 150), (300, 300), (600, 600)]
    
    for width, height in sizes:
        # Redimensionar imagen
        thumbnail = image.copy()
        thumbnail.thumbnail((width, height), Image.Resampling.LANCZOS)
        
        # Convertir a bytes
        output = io.BytesIO()
        thumbnail.save(output, format='JPEG', quality=85)
        thumbnails[f"{width}x{height}"] = output.getvalue()
    
    return thumbnails

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

import json
import boto3
import numpy as np
from sklearn.externals import joblib

# Cargar modelo pre-entrenado
model = joblib.load('/opt/ml/model.pkl')
s3 = boto3.client('s3')

def lambda_handler(event, context):
    """
    API de predicción ML
    """
    
    try:
        # Parsear datos de entrada
        input_data = json.loads(event['body'])
        features = np.array(input_data['features']).reshape(1, -1)
        
        # Hacer predicción
        prediction = model.predict(features)[0]
        probability = model.predict_proba(features)[0].max()
        
        # Guardar predicción para análisis
        save_prediction_log({
            'input': input_data,
            'prediction': float(prediction),
            'probability': float(probability),
            'timestamp': datetime.utcnow().isoformat()
        })
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'prediction': float(prediction),
                'confidence': float(probability)
            })
        }
        
    except Exception as e:
        return {
            'statusCode': 400,
            'body': json.dumps({'error': str(e)})
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:PutItem",
        "dynamodb:UpdateItem"
      ],
      "Resource": "arn:aws:dynamodb:region:account:table/MyTable"
    }
  ]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

import boto3
import json
from botocore.exceptions import ClientError

secrets_client = boto3.client('secretsmanager')

def get_secret(secret_name):
    """
    Obtiene secreto de AWS Secrets Manager
    """
    try:
        response = secrets_client.get_secret_value(SecretId=secret_name)
        return json.loads(response['SecretString'])
    except ClientError as e:
        logger.error(f"Error retrieving secret: {e}")
        raise

def lambda_handler(event, context):
    """
    Handler que usa secretos de forma segura
    """
    
    # Obtener credenciales de base de datos
    db_credentials = get_secret('prod/database/credentials')
    
    # Usar credenciales para conectar
    connection = create_db_connection(
        host=db_credentials['host'],
        username=db_credentials['username'],
        password=db_credentials['password']
    )
    
    # Procesar request
    return process_with_db(event, connection)