fix(data_classes): add AppSyncIdentityLambda for AWS_LAMBDA AppSync 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
+    if "resolverContext" in identity:
+        return AppSyncIdentityLambda(identity)
+
     # AWS_IAM authorization
     return AppSyncIdentityIAM(identity)
 
@@ -54,13 +58,13 @@ 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 identify found:
         - API_KEY authorization - returns None
         - AWS_IAM authorization - returns AppSyncIdentityIAM
         - AMAZON_COGNITO_USER_POOLS authorization - returns AppSyncIdentityCognito
-        - AWS_LAMBDA authorization - returns None - NEED TO TEST
+        - AWS_LAMBDA authorization - returns AppSyncIdentityLambda
         - OPENID_CONNECT authorization - returns None - NEED TO TEST
         """
         return get_identity_object(self.get("identity"))
@@ -150,6 +154,19 @@ def issuer(self) -> str:
         return self["issuer"]
 
 
+class AppSyncIdentityLambda(DictWrapper):
+    """AWS_LAMBDA authorization
+
+    When using a Lambda authorizer, the identity object contains only the
+    ``resolverContext`` returned by the authorizer.
+    """
+
+    @property
+    def resolver_context(self) -> dict[str, Any]:
+        """The resolver context returned by the Lambda authorizer."""
+        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`.
 
+Depending on the authorization type, `AppSyncResolverEvent.identity` returns `AppSyncIdentityCognito` (Cognito User Pools / OpenID Connect), `AppSyncIdentityIAM` (AWS IAM), `AppSyncIdentityLambda` (Lambda authorizer, exposing the authorizer's `resolver_context`), or `None` (API key).
+
 === "app.py"
 
     ```python hl_lines="2-4 9"

tests/events/appSyncResolverEventLambdaAuth.json

@@ -0,0 +1,59 @@
+{
+  "typeName": "Merchant",
+  "fieldName": "locations",
+  "arguments": {
+    "page": 2,
+    "size": 1,
+    "name": "value"
+  },
+  "identity": {
+    "resolverContext": {
+      "userId": "u-1",
+      "tier": "gold",
+      "nested": {
+        "team": "eng"
+      }
+    }
+  },
+  "source": {
+    "name": "Value",
+    "nested": {
+      "name": "value",
+      "list": []
+    }
+  },
+  "request": {
+    "headers": {
+      "x-forwarded-for": "11.215.2.22, 64.44.173.11",
+      "cloudfront-viewer-country": "US",
+      "cloudfront-is-tablet-viewer": "false",
+      "via": "2.0 SOMETHING.cloudfront.net (CloudFront)",
+      "cloudfront-forwarded-proto": "https",
+      "origin": "https://console.aws.amazon.com",
+      "content-length": "156",
+      "accept-language": "en-US,en;q=0.9",
+      "host": "SOMETHING.appsync-api.us-east-1.amazonaws.com",
+      "x-forwarded-proto": "https",
+      "sec-gpc": "1",
+      "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) etc.",
+      "accept": "*/*",
+      "cloudfront-is-mobile-viewer": "false",
+      "cloudfront-is-smarttv-viewer": "false",
+      "accept-encoding": "gzip, deflate, br",
+      "referer": "https://console.aws.amazon.com/",
+      "content-type": "application/json",
+      "sec-fetch-mode": "cors",
+      "x-amz-cf-id": "Fo5VIuvP6V6anIEt62WzFDCK45mzM4yEdpt5BYxOl9OFqafd-WR0cA==",
+      "x-amzn-trace-id": "Root=1-60488877-0b0c4e6727ab2a1c545babd0",
+      "authorization": "AUTH-HEADER",
+      "sec-fetch-dest": "empty",
+      "x-amz-user-agent": "AWS-Console-AppSync/",
+      "cloudfront-is-desktop-viewer": "true",
+      "sec-fetch-site": "cross-site",
+      "x-forwarded-port": "443"
+    }
+  },
+  "prev": {
+    "result": {}
+  }
+}

tests/unit/data_classes/required_dependencies/test_appsync_resolver_event.py

@@ -143,3 +143,55 @@ def test_appsync_resolver_event_empty():
 
     assert event.info.field_name is None
     assert event.info.parent_type_name is None
+
+
+def test_appsync_lambda_identity_class_exposes_resolver_context():
+    # AWS_LAMBDA authorization: the identity carries only what the authorizer returned
+    from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import AppSyncIdentityLambda
+
+    identity = AppSyncIdentityLambda({"resolverContext": {"userId": "u-1", "tier": "gold"}})
+
+    assert identity.resolver_context == {"userId": "u-1", "tier": "gold"}
+
+
+def test_appsync_resolver_event_lambda_identity():
+    # AWS_LAMBDA authorization: identity with resolverContext must dispatch to AppSyncIdentityLambda
+    from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import AppSyncIdentityLambda
+
+    raw_event = load_event("appSyncResolverEventLambdaAuth.json")
+    parsed_event = AppSyncResolverEvent(raw_event)
+
+    assert isinstance(parsed_event.identity, AppSyncIdentityLambda)
+    assert not isinstance(parsed_event.identity, AppSyncIdentityIAM)
+    assert parsed_event.identity.resolver_context == raw_event["identity"]["resolverContext"]
+
+
+def test_get_identity_object_iam_unchanged_without_resolver_context():
+    # AWS_IAM authorization must keep falling through when no resolverContext is present
+    identity = {
+        "accountId": "123456789012",
+        "cognitoIdentityAuthProvider": "cognitoIdentityAuthProvider",
+        "cognitoIdentityAuthType": "cognitoIdentityAuthType",
+        "cognitoIdentityId": "cognitoIdentityId",
+        "cognitoIdentityPoolId": "cognitoIdentityPoolId",
+        "sourceIp": ["11.215.2.22"],
+        "userArn": "userArn",
+        "username": "username",
+    }
+
+    identity_object = get_identity_object(identity)
+
+    assert isinstance(identity_object, AppSyncIdentityIAM)
+    assert identity_object.account_id == identity["accountId"]
+
+
+def test_get_identity_object_lambda_resolver_context_passthrough():
+    # AWS_LAMBDA authorization: resolver_context must pass through nested structures untouched
+    from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import AppSyncIdentityLambda
+
+    resolver_context = {"userId": "u-1", "nested": {"team": "eng", "roles": ["admin"]}}
+    identity_object = get_identity_object({"resolverContext": resolver_context})
+
+    assert isinstance(identity_object, AppSyncIdentityLambda)
+    assert identity_object.resolver_context == resolver_context
+    assert identity_object.resolver_context["nested"]["roles"] == ["admin"]