tessl/pypi-aws-lambda-powertools

Comprehensive developer toolkit implementing serverless best practices for AWS Lambda functions in Python

1.21x

Overview

Eval results

Files

Parser

Name: tessl/pypi-aws-lambda-powertools
Rating: 0.89 (1 reviews)
Author: tessl

Event parsing and validation using Pydantic models with built-in envelopes for extracting business logic from AWS event sources. Enables type-safe event processing and automatic validation of incoming Lambda events.

Capabilities

Event Parser Functions

High-level functions for parsing and validating Lambda events using Pydantic models.

def event_parser(
    model: BaseModel,
    envelope: BaseEnvelope = None,
) -> Callable:
    """
    Decorator for parsing Lambda events into Pydantic models.
    
    Parameters:
    - model: Pydantic model class to parse event into
    - envelope: Envelope to extract data from event structure
    
    Returns:
    Decorated function that receives parsed model instance as parameter
    
    Raises:
    ValidationError: If event validation fails
    """

def parse(
    event: Dict[str, Any],
    model: BaseModel,
    envelope: BaseEnvelope = None,
) -> Any:
    """
    Parse event data into Pydantic model instance.
    
    Parameters:
    - event: Raw Lambda event dictionary
    - model: Pydantic model class for parsing
    - envelope: Optional envelope for data extraction
    
    Returns:
    Parsed model instance
    
    Raises:
    ValidationError: If parsing/validation fails
    """

Base Model and Validation

Pydantic base model and validation utilities re-exported for convenience.

class BaseModel:
    """
    Pydantic BaseModel for defining data schemas.
    Re-exported from pydantic for parser functionality.
    """
    
    def __init__(self, **data: Any): ...
    
    def dict(self, **kwargs) -> Dict[str, Any]:
        """Convert model to dictionary"""
    
    def json(self, **kwargs) -> str:
        """Convert model to JSON string"""
    
    @classmethod
    def parse_obj(cls, obj: Dict[str, Any]) -> "BaseModel":
        """Parse dictionary into model instance"""
    
    @classmethod
    def parse_raw(cls, data: str, **kwargs) -> "BaseModel":
        """Parse raw string data into model instance"""
    
    @classmethod
    def schema(cls, **kwargs) -> Dict[str, Any]:
        """Get JSON schema for model"""

def Field(
    default: Any = ...,
    alias: str = None,
    title: str = None,
    description: str = None,
    gt: float = None,
    ge: float = None,
    lt: float = None,
    le: float = None,
    min_length: int = None,
    max_length: int = None,
    regex: str = None,
    **kwargs,
) -> Any:
    """
    Pydantic Field function for model field configuration.
    
    Parameters:
    - default: Default field value
    - alias: Field alias for serialization
    - title: Field title for schema
    - description: Field description for schema
    - gt: Numeric greater than validation
    - ge: Numeric greater than or equal validation
    - lt: Numeric less than validation
    - le: Numeric less than or equal validation
    - min_length: Minimum string/list length
    - max_length: Maximum string/list length
    - regex: Regular expression pattern validation
    - **kwargs: Additional field options
    
    Returns:
    Field configuration object
    """

def field_validator(field: str, **kwargs) -> Callable:
    """
    Decorator for field-level validation.
    
    Parameters:
    - field: Field name to validate
    - **kwargs: Validator options
    
    Returns:
    Field validator decorator
    """

def model_validator(mode: str = "before", **kwargs) -> Callable:
    """
    Decorator for model-level validation.
    
    Parameters:
    - mode: Validation mode ("before" or "after")
    - **kwargs: Validator options
    
    Returns:
    Model validator decorator
    """

class ValidationError(Exception):
    """
    Pydantic validation error.
    Raised when model validation fails.
    """
    
    def __init__(self, errors: List[Dict[str, Any]]): ...
    
    @property
    def errors(self) -> List[Dict[str, Any]]:
        """Get validation error details"""

Envelope Classes

Envelopes for extracting business data from AWS event structures.

class BaseEnvelope:
    """Base envelope for event data extraction"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        """
        Extract and parse data from event structure.
        
        Parameters:
        - data: Raw event data
        - model: Pydantic model for parsing
        
        Returns:
        Parsed model instance or list of instances
        """

class ApiGatewayEnvelope(BaseEnvelope):
    """Envelope for API Gateway REST API events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        """Extract body from API Gateway event and parse with model"""

class ApiGatewayV2Envelope(BaseEnvelope):
    """Envelope for API Gateway HTTP API (v2.0) events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        """Extract body from HTTP API event and parse with model"""

class ApiGatewayWebSocketEnvelope(BaseEnvelope):
    """Envelope for API Gateway WebSocket events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        """Extract body from WebSocket event and parse with model"""

class LambdaFunctionUrlEnvelope(BaseEnvelope):
    """Envelope for Lambda Function URL events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        """Extract body from Function URL event and parse with model"""

class ALBEnvelope(BaseEnvelope):
    """Envelope for Application Load Balancer events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        """Extract body from ALB event and parse with model"""

class SqsEnvelope(BaseEnvelope):
    """Envelope for SQS events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> List[Any]:
        """Extract and parse each SQS message body with model"""

class SnsEnvelope(BaseEnvelope):
    """Envelope for SNS events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> List[Any]:
        """Extract and parse each SNS message with model"""

class SnsSqsEnvelope(BaseEnvelope):
    """Envelope for SNS messages delivered via SQS"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> List[Any]:
        """Extract SNS messages from SQS records and parse with model"""

class EventBridgeEnvelope(BaseEnvelope):
    """Envelope for EventBridge events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        """Extract detail from EventBridge event and parse with model"""

class CloudWatchLogsEnvelope(BaseEnvelope):
    """Envelope for CloudWatch Logs events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> List[Any]:
        """Extract and parse CloudWatch log events with model"""

class KinesisDataStreamEnvelope(BaseEnvelope):
    """Envelope for Kinesis Data Streams events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> List[Any]:
        """Extract and parse Kinesis records with model"""

class KinesisFirehoseEnvelope(BaseEnvelope):
    """Envelope for Kinesis Data Firehose transformation events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> List[Any]:
        """Extract and parse Firehose records with model"""

class DynamoDBStreamEnvelope(BaseEnvelope):
    """Envelope for DynamoDB Streams events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> List[Any]:
        """Extract and parse DynamoDB Stream records with model"""

class KafkaEnvelope(BaseEnvelope):
    """Envelope for Kafka events (MSK/Self-managed)"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> List[Any]:
        """Extract and parse Kafka records with model"""

class VpcLatticeEnvelope(BaseEnvelope):
    """Envelope for VPC Lattice events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        """Extract body from VPC Lattice event and parse with model"""

class VpcLatticeV2Envelope(BaseEnvelope):
    """Envelope for VPC Lattice V2 events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        """Extract body from VPC Lattice V2 event and parse with model"""

class BedrockAgentEnvelope(BaseEnvelope):
    """Envelope for Bedrock Agent events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        """Extract parameters from Bedrock Agent event and parse with model"""

class BedrockAgentFunctionEnvelope(BaseEnvelope):
    """Envelope for Bedrock Agent Function events"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        """Extract parameters from Bedrock Agent Function event and parse with model"""

Parser Models

Pre-built Pydantic models for common AWS event types and data structures.

# Note: The models module contains 100+ pre-built Pydantic models
# Here are key examples - see full documentation for complete list

class S3Model(BaseModel):
    """S3 object information model"""
    bucket: str = Field(description="S3 bucket name")
    key: str = Field(description="S3 object key")
    size: Optional[int] = Field(description="Object size in bytes")
    etag: Optional[str] = Field(description="Object ETag")

class SqsModel(BaseModel):
    """SQS message model"""
    message_id: str = Field(description="SQS message ID")
    receipt_handle: str = Field(description="SQS receipt handle")
    body: str = Field(description="Message body")
    attributes: Optional[Dict[str, str]] = Field(description="Message attributes")

class DynamoDbModel(BaseModel):
    """DynamoDB item model"""
    keys: Dict[str, Any] = Field(description="Primary key values")
    new_image: Optional[Dict[str, Any]] = Field(description="New item image")
    old_image: Optional[Dict[str, Any]] = Field(description="Old item image")
    event_name: str = Field(description="Event type (INSERT, MODIFY, REMOVE)")

class KinesisModel(BaseModel):
    """Kinesis record model"""
    partition_key: str = Field(description="Partition key")
    sequence_number: str = Field(description="Sequence number")
    data: bytes = Field(description="Record data")
    approximate_arrival_timestamp: Optional[datetime] = Field(description="Arrival timestamp")

class EventBridgeModel(BaseModel):
    """EventBridge event model"""
    source: str = Field(description="Event source")
    detail_type: str = Field(description="Event detail type")
    detail: Dict[str, Any] = Field(description="Event detail")
    resources: Optional[List[str]] = Field(description="Event resources")

Type Definitions

Type aliases and utilities for parser functionality.

from typing import Any, Dict, List, Union, Type

# JSON type alias for any JSON-serializable value
Json = Union[Dict[str, Any], List[Any], str, int, float, bool, None]

# Literal type alias (re-exported from typing_extensions for compatibility)
Literal = Any  # Actual implementation uses typing_extensions.Literal

Usage Examples

Basic Event Parsing

from aws_lambda_powertools.utilities.parser import event_parser, BaseModel, Field
from aws_lambda_powertools.utilities.typing import LambdaContext
from typing import List, Optional

# Define custom data models
class OrderItem(BaseModel):
    product_id: str = Field(description="Product identifier")
    quantity: int = Field(gt=0, description="Item quantity")
    price: float = Field(gt=0, description="Item price")

class Order(BaseModel):
    order_id: str = Field(description="Order identifier")
    customer_id: str = Field(description="Customer identifier") 
    items: List[OrderItem] = Field(description="Order items")
    total: Optional[float] = Field(ge=0, description="Order total")
    status: str = Field(default="pending", description="Order status")

# Parse API Gateway event automatically
@event_parser(model=Order)
def lambda_handler(event: Order, context: LambdaContext) -> dict:
    # Event is automatically parsed and validated
    print(f"Processing order {event.order_id} for customer {event.customer_id}")
    
    # Access validated data
    total_items = sum(item.quantity for item in event.items)
    calculated_total = sum(item.quantity * item.price for item in event.items)
    
    # Validate total if provided
    if event.total and abs(event.total - calculated_total) > 0.01:
        return {
            "statusCode": 400,
            "body": {"error": "Order total mismatch"}
        }
    
    # Process order
    result = process_order(event)
    
    return {
        "statusCode": 200,
        "body": {
            "order_id": event.order_id,
            "total_items": total_items,
            "calculated_total": calculated_total,
            "result": result
        }
    }

def process_order(order: Order) -> dict:
    """Process validated order"""
    # Business logic with type-safe access
    return {
        "processed": True,
        "items_count": len(order.items),
        "customer_id": order.customer_id
    }

SQS Batch Processing with Parser

from aws_lambda_powertools.utilities.parser import event_parser, BaseModel, Field
from aws_lambda_powertools.utilities.parser.envelopes import SqsEnvelope
from aws_lambda_powertools.utilities.typing import LambdaContext
from typing import List
from datetime import datetime

class UserEvent(BaseModel):
    user_id: str = Field(description="User identifier")
    event_type: str = Field(description="Event type")
    timestamp: datetime = Field(description="Event timestamp")
    properties: dict = Field(default_factory=dict, description="Event properties")

# Parse SQS messages into UserEvent models
@event_parser(model=UserEvent, envelope=SqsEnvelope)
def lambda_handler(events: List[UserEvent], context: LambdaContext) -> dict:
    processed_events = []
    
    for event in events:
        # Each SQS message is parsed and validated
        print(f"Processing {event.event_type} event for user {event.user_id}")
        
        # Type-safe access to event data
        result = process_user_event(event)
        processed_events.append(result)
    
    return {
        "statusCode": 200,
        "processed_count": len(processed_events),
        "results": processed_events
    }

def process_user_event(event: UserEvent) -> dict:
    """Process individual user event"""
    if event.event_type == "login":
        return handle_login_event(event)
    elif event.event_type == "purchase":
        return handle_purchase_event(event)
    else:
        return {"status": "unknown_event_type"}

def handle_login_event(event: UserEvent) -> dict:
    """Handle user login event"""
    ip_address = event.properties.get("ip_address")
    device = event.properties.get("device")
    
    return {
        "event_type": "login",
        "user_id": event.user_id,
        "ip_address": ip_address,
        "device": device,
        "processed_at": datetime.utcnow().isoformat()
    }

def handle_purchase_event(event: UserEvent) -> dict:
    """Handle purchase event"""
    amount = event.properties.get("amount", 0)
    currency = event.properties.get("currency", "USD")
    
    return {
        "event_type": "purchase",
        "user_id": event.user_id,
        "amount": amount,
        "currency": currency,
        "processed_at": datetime.utcnow().isoformat()
    }

Custom Validation with Pydantic

from aws_lambda_powertools.utilities.parser import (
    event_parser, 
    BaseModel, 
    Field, 
    field_validator,
    model_validator,
    ValidationError
)
from aws_lambda_powertools.utilities.typing import LambdaContext
from typing import List, Optional
import re

class EmailAddress(BaseModel):
    email: str = Field(description="Email address")
    
    @field_validator('email')
    @classmethod
    def validate_email(cls, v: str) -> str:
        """Validate email format"""
        email_regex = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(email_regex, v):
            raise ValueError("Invalid email format")
        return v.lower()

class CustomerRegistration(BaseModel):
    first_name: str = Field(min_length=1, max_length=50, description="First name")
    last_name: str = Field(min_length=1, max_length=50, description="Last name")
    email: str = Field(description="Email address")
    phone: Optional[str] = Field(default=None, description="Phone number")
    age: int = Field(ge=13, le=120, description="Customer age")
    marketing_consent: bool = Field(default=False, description="Marketing consent")
    
    @field_validator('email')
    @classmethod
    def validate_email(cls, v: str) -> str:
        """Validate email format"""
        email_regex = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(email_regex, v):
            raise ValueError("Invalid email format")
        return v.lower()
    
    @field_validator('phone')
    @classmethod
    def validate_phone(cls, v: Optional[str]) -> Optional[str]:
        """Validate phone number format"""
        if v is None:
            return v
        
        # Remove common separators
        phone_clean = re.sub(r'[\s\-\(\)]+', '', v)
        
        # Check if it's a valid phone number (simple validation)
        if not re.match(r'^\+?[\d]{10,15}$', phone_clean):
            raise ValueError("Invalid phone number format")
        
        return phone_clean
    
    @model_validator(mode='after')
    def validate_marketing_consent(self) -> 'CustomerRegistration':
        """Validate marketing consent rules"""
        # European customers under 16 cannot consent to marketing
        if self.age < 16 and self.marketing_consent:
            raise ValueError("Customers under 16 cannot consent to marketing")
        
        return self

@event_parser(model=CustomerRegistration)
def lambda_handler(event: CustomerRegistration, context: LambdaContext) -> dict:
    try:
        # Event is automatically validated
        print(f"Registering customer: {event.first_name} {event.last_name}")
        
        # Process registration
        customer_id = register_customer(event)
        
        return {
            "statusCode": 200,
            "body": {
                "customer_id": customer_id,
                "message": "Registration successful"
            }
        }
        
    except ValidationError as e:
        # Handle validation errors
        print(f"Validation failed: {e.errors()}")
        return {
            "statusCode": 400,
            "body": {
                "error": "Invalid registration data",
                "details": [
                    {"field": err["loc"][-1], "message": err["msg"]}
                    for err in e.errors()
                ]
            }
        }

def register_customer(registration: CustomerRegistration) -> str:
    """Register new customer with validated data"""
    import uuid
    
    customer_id = str(uuid.uuid4())
    
    # Save to database (mock)
    customer_data = {
        "id": customer_id,
        "first_name": registration.first_name,
        "last_name": registration.last_name,
        "email": registration.email,
        "phone": registration.phone,
        "age": registration.age,
        "marketing_consent": registration.marketing_consent,
        "created_at": datetime.utcnow().isoformat()
    }
    
    print(f"Saving customer: {customer_data}")
    
    return customer_id

Multiple Event Types with Custom Envelopes

from aws_lambda_powertools.utilities.parser import (
    parse,
    BaseModel,
    Field,
    BaseEnvelope
)
from aws_lambda_powertools.utilities.parser.envelopes import (
    SqsEnvelope,
    EventBridgeEnvelope,
    KinesisDataStreamEnvelope
)
from aws_lambda_powertools.utilities.typing import LambdaContext
from typing import List, Union, Any, Dict

# Define different event models
class OrderCreated(BaseModel):
    order_id: str = Field(description="Order ID")
    customer_id: str = Field(description="Customer ID")
    total_amount: float = Field(ge=0, description="Order total")
    currency: str = Field(default="USD", description="Currency code")

class PaymentProcessed(BaseModel):
    payment_id: str = Field(description="Payment ID")
    order_id: str = Field(description="Related order ID")
    amount: float = Field(ge=0, description="Payment amount")
    status: str = Field(description="Payment status")

class InventoryUpdate(BaseModel):
    product_id: str = Field(description="Product ID")
    quantity_change: int = Field(description="Quantity change (+/-)")
    warehouse_id: str = Field(description="Warehouse ID")

def lambda_handler(event: dict, context: LambdaContext) -> dict:
    """Handle multiple event sources and types"""
    
    # Determine event source and parse accordingly
    if "Records" in event:
        if event["Records"][0].get("eventSource") == "aws:sqs":
            return handle_sqs_events(event, context)
        elif event["Records"][0].get("eventSource") == "aws:kinesis":
            return handle_kinesis_events(event, context)
    elif "source" in event:
        return handle_eventbridge_event(event, context)
    else:
        return {"statusCode": 400, "body": "Unknown event type"}

def handle_sqs_events(event: dict, context: LambdaContext) -> dict:
    """Handle SQS events with different message types"""
    results = []
    
    # Parse SQS envelope to get message bodies
    for record in event["Records"]:
        message_body = record["body"]
        
        # Parse message to determine type
        try:
            import json
            message_data = json.loads(message_body)
            event_type = message_data.get("type")
            
            if event_type == "order_created":
                parsed_event = parse(message_data["data"], OrderCreated)
                result = process_order_created(parsed_event)
            elif event_type == "payment_processed":
                parsed_event = parse(message_data["data"], PaymentProcessed)
                result = process_payment(parsed_event)
            else:
                result = {"error": f"Unknown event type: {event_type}"}
            
            results.append(result)
            
        except Exception as e:
            results.append({"error": str(e)})
    
    return {
        "statusCode": 200,
        "processed": len(results),
        "results": results
    }

def handle_eventbridge_event(event: dict, context: LambdaContext) -> dict:
    """Handle EventBridge events"""
    
    detail_type = event.get("detail-type")
    
    try:
        if detail_type == "Order Created":
            parsed_event = parse(event, OrderCreated, envelope=EventBridgeEnvelope)
            result = process_order_created(parsed_event)
        elif detail_type == "Payment Processed":
            parsed_event = parse(event, PaymentProcessed, envelope=EventBridgeEnvelope)
            result = process_payment(parsed_event)
        else:
            result = {"error": f"Unknown detail type: {detail_type}"}
        
        return {
            "statusCode": 200,
            "result": result
        }
        
    except Exception as e:
        return {
            "statusCode": 500,
            "error": str(e)
        }

def handle_kinesis_events(event: dict, context: LambdaContext) -> dict:
    """Handle Kinesis stream events"""
    
    try:
        # Parse all Kinesis records as inventory updates
        inventory_updates = parse(event, InventoryUpdate, envelope=KinesisDataStreamEnvelope)
        
        results = []
        for update in inventory_updates:
            result = process_inventory_update(update)
            results.append(result)
        
        return {
            "statusCode": 200,
            "processed": len(results),
            "results": results
        }
        
    except Exception as e:
        return {
            "statusCode": 500,
            "error": str(e)
        }

def process_order_created(order: OrderCreated) -> dict:
    """Process order creation event"""
    return {
        "type": "order_processed",
        "order_id": order.order_id,
        "customer_id": order.customer_id,
        "amount": order.total_amount,
        "currency": order.currency
    }

def process_payment(payment: PaymentProcessed) -> dict:
    """Process payment event"""
    return {
        "type": "payment_processed",
        "payment_id": payment.payment_id,
        "order_id": payment.order_id,
        "amount": payment.amount,
        "status": payment.status
    }

def process_inventory_update(update: InventoryUpdate) -> dict:
    """Process inventory update event"""
    return {
        "type": "inventory_updated",
        "product_id": update.product_id,
        "quantity_change": update.quantity_change,
        "warehouse_id": update.warehouse_id
    }

Advanced Parser Features

from aws_lambda_powertools.utilities.parser import (
    event_parser,
    BaseModel,
    Field,
    field_validator,
    model_validator,
    BaseEnvelope
)
from aws_lambda_powertools.utilities.typing import LambdaContext
from typing import List, Optional, Union, Any
from datetime import datetime, timezone
from decimal import Decimal
import json

class CustomEnvelope(BaseEnvelope):
    """Custom envelope for proprietary event format"""
    
    def parse(self, data: Dict[str, Any], model: BaseModel) -> Any:
        # Extract nested payload from custom format
        payload = data.get("payload", {})
        metadata = data.get("metadata", {})
        
        # Merge metadata into payload for parsing
        if metadata:
            payload["_metadata"] = metadata
        
        return model.parse_obj(payload)

class Money(BaseModel):
    """Money value with currency"""
    amount: Decimal = Field(description="Monetary amount")
    currency: str = Field(default="USD", description="Currency code")
    
    @field_validator('currency')
    @classmethod
    def validate_currency(cls, v: str) -> str:
        """Validate currency code"""
        valid_currencies = ["USD", "EUR", "GBP", "JPY", "CAD", "AUD"]
        if v.upper() not in valid_currencies:
            raise ValueError(f"Invalid currency: {v}")
        return v.upper()

class Address(BaseModel):
    """Address information"""
    street: str = Field(description="Street address")
    city: str = Field(description="City")
    state: Optional[str] = Field(default=None, description="State/Province")
    postal_code: str = Field(description="Postal/ZIP code")
    country: str = Field(description="Country code")

class ComplexOrder(BaseModel):
    """Complex order with nested models and validation"""
    order_id: str = Field(description="Order identifier")
    customer_id: str = Field(description="Customer identifier")
    order_date: datetime = Field(description="Order date")
    
    # Nested models
    billing_address: Address = Field(description="Billing address")
    shipping_address: Optional[Address] = Field(default=None, description="Shipping address")
    
    # Complex fields
    total: Money = Field(description="Order total")
    tax: Money = Field(description="Tax amount")
    shipping_cost: Money = Field(description="Shipping cost")
    
    # Optional metadata
    _metadata: Optional[dict] = Field(default=None, alias="metadata", description="Event metadata")
    
    @field_validator('order_date')
    @classmethod
    def validate_order_date(cls, v: datetime) -> datetime:
        """Ensure order date is timezone-aware"""
        if v.tzinfo is None:
            v = v.replace(tzinfo=timezone.utc)
        return v
    
    @model_validator(mode='after')
    def validate_order_totals(self) -> 'ComplexOrder':
        """Validate order financial calculations"""
        # Ensure all amounts use same currency
        currencies = {self.total.currency, self.tax.currency, self.shipping_cost.currency}
        if len(currencies) > 1:
            raise ValueError("All monetary amounts must use the same currency")
        
        # Validate total calculation (simplified)
        calculated_total = self.tax.amount + self.shipping_cost.amount
        if abs(self.total.amount - calculated_total) > Decimal('0.01'):
            # In a real scenario, you'd validate against line items too
            pass  # Skip validation for this example
        
        return self
    
    @model_validator(mode='after')
    def set_default_shipping_address(self) -> 'ComplexOrder':
        """Set shipping address to billing if not provided"""
        if self.shipping_address is None:
            self.shipping_address = self.billing_address
        
        return self

@event_parser(model=ComplexOrder, envelope=CustomEnvelope)
def lambda_handler(event: ComplexOrder, context: LambdaContext) -> dict:
    """Process complex order with full validation"""
    
    print(f"Processing order {event.order_id} for customer {event.customer_id}")
    
    # Access nested model data
    billing_city = event.billing_address.city
    shipping_city = event.shipping_address.city
    
    # Work with validated monetary amounts
    total_amount = event.total.amount
    currency = event.total.currency
    
    # Access metadata if provided
    source_system = None
    if event._metadata:
        source_system = event._metadata.get("source_system")
    
    # Process order
    result = {
        "order_id": event.order_id,
        "customer_id": event.customer_id,
        "total_amount": str(total_amount),
        "currency": currency,
        "billing_city": billing_city,
        "shipping_city": shipping_city,
        "source_system": source_system,
        "processed_at": datetime.utcnow().isoformat()
    }
    
    # Perform business logic
    process_complex_order(event)
    
    return {
        "statusCode": 200,
        "body": result
    }

def process_complex_order(order: ComplexOrder):
    """Process order with type-safe access to all fields"""
    
    # Calculate shipping distance (mock)
    if order.billing_address.city != order.shipping_address.city:
        print(f"Cross-city shipping: {order.billing_address.city} -> {order.shipping_address.city}")
    
    # Handle international orders
    if order.billing_address.country != order.shipping_address.country:
        print(f"International order: {order.billing_address.country} -> {order.shipping_address.country}")
        # Apply international shipping rules
    
    # Process payment
    print(f"Processing payment: {order.total.amount} {order.total.currency}")
    
    # Log order details
    print(f"Order details: {order.dict()}")

Types

from typing import Dict, Any, List, Union, Type, Optional, Callable
from pydantic import BaseModel as PydanticBaseModel, ValidationError as PydanticValidationError

# Parser function signatures
EventParserDecorator = Callable[[Callable], Callable]
ParseFunction = Callable[[Dict[str, Any], Type[BaseModel], Optional[BaseEnvelope]], Any]

# Model types
ModelClass = Type[BaseModel]
ModelInstance = BaseModel

# Validation types
ValidationError = PydanticValidationError
ValidationErrors = List[Dict[str, Any]]

# Envelope types
EnvelopeClass = Type[BaseEnvelope]
EnvelopeInstance = BaseEnvelope

# Event data types
EventData = Dict[str, Any]
ParsedData = Union[Any, List[Any]]

# Field validation decorator
FieldValidator = Callable[[Any], Any]
ModelValidator = Callable[[BaseModel], BaseModel]

# JSON types
JsonValue = Union[str, int, float, bool, None, Dict[str, Any], List[Any]]
Json = JsonValue

# Literal type for type hints
from typing_extensions import Literal

Install with Tessl CLI

npx tessl i tessl/pypi-aws-lambda-powertools