fix(data_classes): add AppSyncIdentityLambda for AWS_LAMBDA authorization

aws_lambda_powertools/utilities/data_classes/appsync_resolver_event.py

@@ -22,6 +22,10 @@ def get_identity_object(identity: dict | None) -> Any:
     if "sub" in identity:
         return AppSyncIdentityCognito(identity)
 
+    # AWS_LAMBDA authorization - the identity is whatever the authorizer returned in resolverContext
+    if "resolverContext" in identity:
+        return AppSyncIdentityLambda(identity)
+
     # AWS_IAM authorization
     return AppSyncIdentityIAM(identity)
 
@@ -54,14 +58,14 @@ def stash(self) -> dict:
         return self.get("stash") or {}
 
     @property
-    def identity(self) -> AppSyncIdentityIAM | AppSyncIdentityCognito | None:
+    def identity(self) -> AppSyncIdentityIAM | AppSyncIdentityCognito | AppSyncIdentityLambda | None:
         """An object that contains information about the caller.
         Depending on the type of identity found:
         - API_KEY authorization - returns None
         - AWS_IAM authorization - returns AppSyncIdentityIAM
         - AMAZON_COGNITO_USER_POOLS authorization - returns AppSyncIdentityCognito
         - OPENID_CONNECT authorization - returns AppSyncIdentityCognito
-        - AWS_LAMBDA authorization - unverified, the mapping is whatever the authorizer returns
+        - AWS_LAMBDA authorization - returns AppSyncIdentityLambda
         """
         return get_identity_object(self.get("identity"))
 
@@ -150,6 +154,15 @@ def issuer(self) -> str:
         return self["issuer"]
 
 
+class AppSyncIdentityLambda(DictWrapper):
+    """AWS_LAMBDA authorization"""
+
+    @property
+    def resolver_context(self) -> dict[str, Any]:
+        """The context that the Lambda authorizer returned under `resolverContext`."""
+        return self["resolverContext"]
+
+
 class AppSyncResolverEventInfo(DictWrapper):
     """The info section contains information about the GraphQL request"""
 

docs/utilities/data_classes.md

@@ -259,6 +259,8 @@ and [AppSync Direct Lambda Resolvers](https://aws.amazon.com/blogs/mobile/appsyn
 
 The example serves as an AppSync resolver for the `locations` field of the `Merchant` type. It uses the `@event_source` decorator to parse the AppSync event, handles pagination and filtering for locations, and demonstrates `AppSyncIdentityCognito`.
 
+With [AWS_LAMBDA authorization](https://docs.aws.amazon.com/appsync/latest/devguide/security-authz.html#aws-lambda-authorization){target="_blank"}, the caller identity is whatever the Lambda authorizer returned in `resolverContext`. In that case, `AppSyncResolverEvent.identity` returns an `AppSyncIdentityLambda`, and the authorizer context is available via `identity.resolver_context`.
+
 === "app.py"
 
     ```python hl_lines="2-4 9"

tests/events/appSyncLambdaAuthorizerResolverEvent.json

@@ -0,0 +1,30 @@
+{
+  "typeName": "Query",
+  "fieldName": "getUser",
+  "arguments": {
+    "userId": "u-1"
+  },
+  "identity": {
+    "resolverContext": {
+      "userId": "u-1",
+      "tier": "gold"
+    }
+  },
+  "source": null,
+  "request": {
+    "headers": {
+      "x-forwarded-for": "11.215.2.22",
+      "host": "SOMETHING.appsync-api.us-east-1.amazonaws.com"
+    },
+    "domainName": "SOMETHING.appsync-api.us-east-1.amazonaws.com"
+  },
+  "prev": null,
+  "info": {
+    "selectionSetList": ["userId", "tier"],
+    "selectionSetGraphQL": "{\n  userId\n  tier\n}",
+    "parentTypeName": "Query",
+    "fieldName": "getUser",
+    "variables": {}
+  },
+  "stash": {}
+}

tests/unit/data_classes/required_dependencies/test_appsync_resolver_event.py

@@ -6,6 +6,7 @@
 from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import (
     AppSyncIdentityCognito,
     AppSyncIdentityIAM,
+    AppSyncIdentityLambda,
     AppSyncResolverEventInfo,
     get_identity_object,
 )
@@ -177,3 +178,37 @@ def test_get_identity_object_only_none_identity_returns_none():
 
     # AND only an absent identity maps to None
     assert get_identity_object(None) is None
+
+
+def test_get_identity_object_lambda_authorizer():
+    # GIVEN an AWS_LAMBDA authorization identity, whose only field is the authorizer's resolverContext
+    identity = {"resolverContext": {"userId": "u-1", "tier": "gold"}}
+
+    # WHEN dispatched to a type
+    identity_object = get_identity_object(identity)
+
+    # THEN it resolves to AppSyncIdentityLambda, exposing the authorizer context
+    assert isinstance(identity_object, AppSyncIdentityLambda)
+    assert identity_object.resolver_context == identity["resolverContext"]
+
+
+def test_get_identity_object_prefers_sub_over_resolver_context():
+    # GIVEN an identity carrying both `sub` and `resolverContext` (authorization modes are exclusive in practice)
+    identity = {"sub": "string", "resolverContext": {"userId": "u-1"}}
+
+    # THEN the existing Cognito dispatch wins, keeping the old behaviour
+    assert isinstance(get_identity_object(identity), AppSyncIdentityCognito)
+
+
+def test_appsync_resolver_event_lambda_authorizer():
+    raw_event = load_event("appSyncLambdaAuthorizerResolverEvent.json")
+    parsed_event = AppSyncResolverEvent(raw_event)
+
+    assert parsed_event.type_name == raw_event["typeName"]
+    assert parsed_event.field_name == raw_event["fieldName"]
+    assert parsed_event.arguments == raw_event["arguments"]
+
+    assert isinstance(parsed_event.identity, AppSyncIdentityLambda)
+    identity: AppSyncIdentityLambda = parsed_event.identity
+    assert identity.resolver_context == raw_event["identity"]["resolverContext"]
+    assert identity.resolver_context["userId"] == "u-1"