|
| 1 | +import base64 |
| 2 | +import binascii |
| 3 | +import json |
| 4 | +from dataclasses import asdict, dataclass, field, is_dataclass |
| 5 | +from uuid import uuid4 |
| 6 | + |
| 7 | +import powertools_base64_jmespath_schema as schemas |
| 8 | +from jmespath.exceptions import JMESPathTypeError |
| 9 | + |
| 10 | +from aws_lambda_powertools.utilities.typing import LambdaContext |
| 11 | +from aws_lambda_powertools.utilities.validation import SchemaValidationError, validate |
| 12 | + |
| 13 | + |
| 14 | +@dataclass |
| 15 | +class Order: |
| 16 | + user_id: int |
| 17 | + product_id: int |
| 18 | + quantity: int |
| 19 | + price: float |
| 20 | + currency: str |
| 21 | + order_id: str = field(default_factory=lambda: f"{uuid4()}") |
| 22 | + |
| 23 | + |
| 24 | +class DataclassCustomEncoder(json.JSONEncoder): |
| 25 | + """A custom JSON encoder to serialize dataclass obj""" |
| 26 | + |
| 27 | + def default(self, obj): |
| 28 | + # Only called for values that aren't JSON serializable |
| 29 | + # where `obj` will be an instance of Todo in this example |
| 30 | + return asdict(obj) if is_dataclass(obj) else super().default(obj) |
| 31 | + |
| 32 | + |
| 33 | +def lambda_handler(event, context: LambdaContext) -> dict: |
| 34 | + |
| 35 | + # Try to validate the schema |
| 36 | + try: |
| 37 | + validate(event=event, schema=schemas.INPUT, envelope="powertools_json(powertools_base64(payload))") |
| 38 | + |
| 39 | + # alternatively, extract_data_from_envelope works here too |
| 40 | + payload_decoded = base64.b64decode(event["payload"]).decode() |
| 41 | + |
| 42 | + order_payload: dict = json.loads(payload_decoded) |
| 43 | + |
| 44 | + return { |
| 45 | + "order": json.dumps(Order(**order_payload), cls=DataclassCustomEncoder), |
| 46 | + "message": "order created", |
| 47 | + "success": True, |
| 48 | + } |
| 49 | + except JMESPathTypeError: |
| 50 | + return return_error_message( |
| 51 | + "The powertools_json(powertools_base64()) envelope function must match a valid path." |
| 52 | + ) |
| 53 | + except binascii.Error: |
| 54 | + return return_error_message("Payload must be a valid base64 encoded string") |
| 55 | + except json.JSONDecodeError: |
| 56 | + return return_error_message("Payload must be valid JSON (base64 encoded).") |
| 57 | + except SchemaValidationError as exception: |
| 58 | + # SchemaValidationError indicates where a data mismatch is |
| 59 | + return return_error_message(str(exception)) |
| 60 | + |
| 61 | + |
| 62 | +def return_error_message(message: str) -> dict: |
| 63 | + return {"order": None, "message": message, "success": False} |
0 commit comments