feat(data-classes): decode base64 encoded body

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"]

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"))