fix(event-sources): Pass authorizer data to APIGatewayEventAuthorizer

Changes:
- contract APIGatewayEventAuthorizer from the authorizer data
- add test cases for the additional fields that can optional be in the authorizer dict
- Update DictWrapper to allow for a default in `get`

Author: Michael Brewer

aws_lambda_powertools/utilities/data_classes/api_gateway_proxy_event.py

@@ -11,11 +11,11 @@
 class APIGatewayEventAuthorizer(DictWrapper):
     @property
     def claims(self) -> Optional[Dict[str, Any]]:
-        return self["requestContext"]["authorizer"].get("claims")
+        return self.get("claims")
 
     @property
     def scopes(self) -> Optional[List[str]]:
-        return self["requestContext"]["authorizer"].get("scopes")
+        return self.get("scopes")
 
 
 class APIGatewayEventRequestContext(BaseRequestContext):
@@ -56,7 +56,7 @@ def route_key(self) -> Optional[str]:
 
     @property
     def authorizer(self) -> APIGatewayEventAuthorizer:
-        return APIGatewayEventAuthorizer(self._data)
+        return APIGatewayEventAuthorizer(self._data["requestContext"]["authorizer"])
 
 
 class APIGatewayProxyEvent(BaseProxyEvent):

aws_lambda_powertools/utilities/data_classes/common.py

@@ -18,8 +18,8 @@ def __eq__(self, other: Any) -> bool:
 
         return self._data == other._data
 
-    def get(self, key: str) -> Optional[Any]:
-        return self._data.get(key)
+    def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
+        return self._data.get(key, default)
 
     @property
     def raw_event(self) -> Dict[str, Any]:

tests/events/apiGatewayProxyEventPrincipalId.json

@@ -0,0 +1,13 @@
+{
+  "resource": "/trip",
+  "path": "/trip",
+  "httpMethod": "POST",
+  "requestContext": {
+    "requestId": "34972478-2843-4ced-a657-253108738274",
+    "authorizer": {
+      "user_id": "fake_username",
+      "principalId": "fake",
+      "integrationLatency": 451
+    }
+  }
+}

tests/functional/test_data_classes.py

@@ -897,6 +897,18 @@ def test_api_gateway_proxy_event():
     assert request_context.identity.client_cert.subject_dn == "www.example.com"
 
 
+def test_api_gateway_proxy_event_with_principal_id():
+    event = APIGatewayProxyEvent(load_event("apiGatewayProxyEventPrincipalId.json"))
+
+    request_context = event.request_context
+    authorizer = request_context.authorizer
+    assert authorizer.claims is None
+    assert authorizer.scopes is None
+    assert authorizer["principalId"] == "fake"
+    assert authorizer.get("principalId") == "fake"
+    assert authorizer.get("foo", "default") == "default"
+
+
 def test_api_gateway_proxy_v2_event():
     event = APIGatewayProxyEventV2(load_event("apiGatewayProxyV2Event.json"))