diff --git a/aws_lambda_powertools/utilities/data_classes/common.py b/aws_lambda_powertools/utilities/data_classes/common.py index a6b975c6072..66c8f15324f 100644 --- a/aws_lambda_powertools/utilities/data_classes/common.py +++ b/aws_lambda_powertools/utilities/data_classes/common.py @@ -1,3 +1,4 @@ +import base64 import json from typing import Any, Dict, Optional @@ -66,6 +67,14 @@ def json_body(self) -> Any: """Parses the submitted body as json""" return json.loads(self["body"]) + @property + def decoded_body(self) -> str: + """Dynamically base64 decode body as a str""" + body: str = self["body"] + if self.is_base64_encoded: + return base64.b64decode(body.encode()).decode() + return body + @property def path(self) -> str: return self["path"] diff --git a/tests/functional/test_data_classes.py b/tests/functional/test_data_classes.py index 8a18df0e0f8..f56d0700e6f 100644 --- a/tests/functional/test_data_classes.py +++ b/tests/functional/test_data_classes.py @@ -5,6 +5,7 @@ from secrets import compare_digest from urllib.parse import quote_plus +import pytest from pytest_mock import MockerFixture from aws_lambda_powertools.utilities.data_classes import ( @@ -848,6 +849,39 @@ def test_base_proxy_event_get_header_value_case_insensitive(): assert value is None +def test_base_proxy_event_json_body_key_error(): + event = BaseProxyEvent({}) + with pytest.raises(KeyError) as ke: + assert not event.json_body + assert str(ke.value) == "'body'" + + +def test_base_proxy_event_json_body(): + data = {"message": "Foo"} + event = BaseProxyEvent({"body": json.dumps(data)}) + assert event.json_body == data + + +def test_base_proxy_event_decode_body_key_error(): + event = BaseProxyEvent({}) + with pytest.raises(KeyError) as ke: + assert not event.decoded_body + assert str(ke.value) == "'body'" + + +def test_base_proxy_event_decode_body_encoded_false(): + data = "Foo" + event = BaseProxyEvent({"body": data, "isBase64Encoded": False}) + assert event.decoded_body == data + + +def test_base_proxy_event_decode_body_encoded_true(): + data = "Foo" + encoded_data = base64.b64encode(data.encode()).decode() + event = BaseProxyEvent({"body": encoded_data, "isBase64Encoded": True}) + assert event.decoded_body == data + + def test_kinesis_stream_event(): event = KinesisStreamEvent(load_event("kinesisStreamEvent.json"))