Sign in Get started

Python Serverless on AWS Lambda · APIs and Events in Python

Validating Events with Pydantic

Why malformed events cause KeyErrors deep in business logic, and how to validate at the boundary.

Scenario

Example app: a function consumes OrderPlaced events from EventBridge and computes loyalty points. Several teams publish events in slightly different shapes.

What you see in production

Errors such as KeyError: 'currency' and TypeError: can't multiply sequence by non-int of type 'float' appear in logs, deep inside the points calculation, for a few percent of events. Some events with a negative quantity are processed and award negative points.

The naive code

def handler(event, context):
    detail = event["detail"]
    points = 0
    for line in detail["lines"]:
        points += line["qty"] * line["unit_price"] * RATE[detail["currency"]]
    award(detail["customer_id"], points)

Reproduce it

Failing deep inside versus rejecting at the boundaryTry it yourself
from dataclasses import dataclass

RATE = {"EUR": 1.0, "USD": 0.9}
events = [
    {"customer_id": "c1", "currency": "EUR", "lines": [{"qty": 2, "unit_price": 10.0}]},
    {"customer_id": "c2", "lines": [{"qty": 1, "unit_price": 5.0}]},                       # no currency
    {"customer_id": "c3", "currency": "USD", "lines": [{"qty": "3", "unit_price": 4.0}]},   # qty as string
    {"customer_id": "c4", "currency": "EUR", "lines": [{"qty": -5, "unit_price": 10.0}]},   # negative
]

def naive(d):
    return sum(l["qty"] * l["unit_price"] * RATE[d["currency"]] for l in d["lines"])

@dataclass
class Line:
    qty: int
    unit_price: float
    def __post_init__(self):
        if not isinstance(self.qty, int) or self.qty <= 0:
            raise ValueError(f"qty must be a positive integer, got {self.qty!r}")

def validate(d):
    if d.get("currency") not in RATE:
        raise ValueError(f"unsupported currency {d.get('currency')!r}")
    return d["customer_id"], d["currency"], [Line(**l) for l in d["lines"]]

for e in events:
    try:
        naive_result = naive(e)
    except Exception as err:
        naive_result = f"{type(err).__name__}: {err}"
    try:
        _, cur, lines = validate(e)
        checked = sum(l.qty * l.unit_price * RATE[cur] for l in lines)
    except ValueError as err:
        checked = f"rejected: {err}"
    print(e["customer_id"], "| naive:", naive_result, "| validated:", checked)

Output

c1 | naive: 20.0 | validated: 20.0
c2 | naive: KeyError: 'currency' | validated: rejected: unsupported currency None
c3 | naive: TypeError: can't multiply sequence by non-int of type 'float' | validated: rejected: qty must be a positive integer, got '3'
c4 | naive: -50.0 | validated: rejected: qty must be a positive integer, got -5

Root cause

Events from other systems are untrusted input. Without validation at the boundary, malformed or missing fields surface as unrelated errors deep in the code, some invalid values (negative quantities) pass silently, and failures are hard to diagnose because the error says nothing about the input contract.

The fix

Define the event contract as a Pydantic model and validate as the first step, with Powertools Parser handling the event envelope (EventBridge, SQS, API Gateway):

from typing import Literal
from pydantic import BaseModel, PositiveInt, PositiveFloat
from aws_lambda_powertools.utilities.parser import event_parser, envelopes

class Line(BaseModel):
    qty: PositiveInt
    unit_price: PositiveFloat

class OrderPlaced(BaseModel):
    customer_id: str
    currency: Literal["EUR", "USD"]
    lines: list[Line]

@event_parser(model=OrderPlaced, envelope=envelopes.EventBridgeEnvelope)
def handler(order: OrderPlaced, context):
    points = sum(l.qty * l.unit_price * RATE[order.currency] for l in order.lines)
    award(order.customer_id, points)

Invalid events raise a ValidationError with field-level details. For asynchronous sources, let them fail into a DLQ (or log and skip, deliberately), and alert the publishing team.

Trade-offs

  • Pydantic adds an import cost at cold start (it uses a compiled core); usually worth it for correctness.
  • Strict models can reject events from publishers that add fields; allow extra fields unless you need to forbid them.
  • Shared models in a package keep publishers and consumers in sync, but couple their release cycles.

Detect it in production

  • Count validation failures per event source and field; alert on spikes after publisher deployments.
  • DLQ contents grouped by validation error.
  • Contract tests with sample events from each publisher.

Key points

  • Treat events as untrusted input; validate at the boundary.
  • Use Pydantic models with Powertools Parser envelopes.
  • Route invalid events to a DLQ or explicit rejection path, with alerts.

Further reading: Powertools for AWS Lambda (Python): Parser · Pydantic documentation

Practice questions

Why validate event payloads at the start of a handler rather than letting errors surface naturally?

Answer

Errors deep in business logic (KeyError, TypeError) hide the real cause, and some invalid values pass silently and corrupt data. Validating at the boundary rejects bad input early with clear, field-level messages.

What does a Powertools Parser envelope do?

Answer

It extracts and validates the payload from the event wrapper of a specific source (for example EventBridge detail, SQS message bodies or API Gateway body) and parses it into your Pydantic model.

What should happen to an asynchronous event that fails validation?

Answer

It should not be retried endlessly: send it to a DLQ or a dedicated error path with the validation details, alert the owners of the publisher, and reprocess it after the publisher or consumer is fixed.

Open in the app to mark lessons complete, take notes and track your progress.