feat(event-handler): add appsync batch resolvers

aws_lambda_powertools/event_handler/appsync.py

@@ -1,5 +1,6 @@
 import logging
-from typing import Any, Callable, Optional, Type, TypeVar
+from itertools import groupby
+from typing import Any, Callable, List, Optional, Type, TypeVar, Union
 
 from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
 from aws_lambda_powertools.utilities.typing import LambdaContext
@@ -10,7 +11,7 @@
 
 
 class BaseRouter:
-    current_event: AppSyncResolverEventT  # type: ignore[valid-type]
+    current_event: Union[AppSyncResolverEventT, List[AppSyncResolverEventT]]  # type: ignore[valid-type]
     lambda_context: LambdaContext
     context: dict
 
@@ -152,11 +153,28 @@ def lambda_handler(event, context):
             If we could not find a field resolver
         """
         # Maintenance: revisit generics/overload to fix [attr-defined] in mypy usage
-        BaseRouter.current_event = data_model(event)
+
         BaseRouter.lambda_context = context
 
-        resolver = self._get_resolver(BaseRouter.current_event.type_name, BaseRouter.current_event.field_name)
-        response = resolver(**BaseRouter.current_event.arguments)
+        # If event is a list it means that AppSync sent batch request
+        if isinstance(event, list):
+            event_groups = [
+                {"field_name": field_name, "events": list(events)}
+                for field_name, events in groupby(event, key=lambda x: x["info"]["fieldName"])
+            ]
+            if len(event_groups) > 1:
+                ValueError("batch with different field names. It shouldn't happen!")
+
+            appconfig_events = [data_model(event) for event in event_groups[0]["events"]]
+            BaseRouter.current_event = appconfig_events
+            resolver = self._get_resolver(appconfig_events[0].type_name, event_groups[0]["field_name"])
+            response = resolver()
+        else:
+            appconfig_event = data_model(event)
+            BaseRouter.current_event = appconfig_event
+            resolver = self._get_resolver(appconfig_event.type_name, appconfig_event.field_name)
+            response = resolver(**appconfig_event.arguments)
+
         self.clear_context()
 
         return response

pyproject.toml

@@ -104,6 +104,7 @@ cfn-lint = "0.74.1"
 mypy = "^0.982"
 types-python-dateutil = "^2.8.19.6"
 httpx = "^0.23.3"
+aws-cdk-aws-appsync-alpha = "^2.59.0a0"
 
 [tool.coverage.run]
 source = ["aws_lambda_powertools"]

tests/e2e/event_handler/files/schema.graphql

@@ -0,0 +1,19 @@
+schema {
+    query: Query
+}
+
+type Query {
+    getPost(post_id:ID!): Post
+    allPosts: [Post]
+}
+
+type Post {
+    post_id: ID!
+    author: String!
+    title: String
+    content: String
+    url: String
+    ups: Int
+    downs: Int
+    relatedPosts: [Post]
+}

tests/e2e/event_handler/handlers/appsync_resolver_handler.py

@@ -0,0 +1,99 @@
+from typing import List
+
+from pydantic import BaseModel
+
+from aws_lambda_powertools.event_handler import AppSyncResolver
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+app = AppSyncResolver()
+
+
+posts = {
+    "1": {
+        "post_id": "1",
+        "title": "First book",
+        "author": "Author1",
+        "url": "https://amazon.com/",
+        "content": "SAMPLE TEXT AUTHOR 1",
+        "ups": "100",
+        "downs": "10",
+    },
+    "2": {
+        "post_id": "2",
+        "title": "Second book",
+        "author": "Author2",
+        "url": "https://amazon.com",
+        "content": "SAMPLE TEXT AUTHOR 2",
+        "ups": "100",
+        "downs": "10",
+    },
+    "3": {
+        "post_id": "3",
+        "title": "Third book",
+        "author": "Author3",
+        "url": None,
+        "content": None,
+        "ups": None,
+        "downs": None,
+    },
+    "4": {
+        "post_id": "4",
+        "title": "Fourth book",
+        "author": "Author4",
+        "url": "https://www.amazon.com/",
+        "content": "SAMPLE TEXT AUTHOR 4",
+        "ups": "1000",
+        "downs": "0",
+    },
+    "5": {
+        "post_id": "5",
+        "title": "Fifth book",
+        "author": "Author5",
+        "url": "https://www.amazon.com/",
+        "content": "SAMPLE TEXT AUTHOR 5",
+        "ups": "50",
+        "downs": "0",
+    },
+}
+
+posts_related = {
+    "1": [posts["4"]],
+    "2": [posts["3"], posts["5"]],
+    "3": [posts["2"], posts["1"]],
+    "4": [posts["2"], posts["1"]],
+    "5": [],
+}
+
+
+class Post(BaseModel):
+    post_id: str
+    author: str
+    title: str
+    url: str
+    content: str
+    ups: str
+    downs: str
+
+
+@app.resolver(type_name="Query", field_name="getPost")
+def get_post(post_id: str = "") -> dict:
+    post = Post(**posts[post_id]).dict()
+    return post
+
+
+@app.resolver(type_name="Query", field_name="allPosts")
+def all_posts() -> List[dict]:
+    return list(posts.values())
+
+
+@app.resolver(type_name="Post", field_name="relatedPosts")
+def related_posts() -> List[dict]:
+    posts = []
+    for resolver_event in app.current_event:
+        if resolver_event.source:
+            posts.append(posts_related[resolver_event.source["post_id"]])
+    return posts
+
+
+def lambda_handler(event, context: LambdaContext) -> dict:
+    return app.resolve(event, context)

tests/e2e/event_handler/infrastructure.py

@@ -1,10 +1,12 @@
+from pathlib import Path
 from typing import Dict, Optional
 
-from aws_cdk import CfnOutput
+from aws_cdk import CfnOutput, Duration, Expiration
 from aws_cdk import aws_apigateway as apigwv1
 from aws_cdk import aws_apigatewayv2_alpha as apigwv2
 from aws_cdk import aws_apigatewayv2_authorizers_alpha as apigwv2authorizers
 from aws_cdk import aws_apigatewayv2_integrations_alpha as apigwv2integrations
+from aws_cdk import aws_appsync_alpha as appsync
 from aws_cdk import aws_ec2 as ec2
 from aws_cdk import aws_elasticloadbalancingv2 as elbv2
 from aws_cdk import aws_elasticloadbalancingv2_targets as targets
@@ -21,6 +23,7 @@ def create_resources(self):
         self._create_api_gateway_rest(function=functions["ApiGatewayRestHandler"])
         self._create_api_gateway_http(function=functions["ApiGatewayHttpHandler"])
         self._create_lambda_function_url(function=functions["LambdaFunctionUrlHandler"])
+        self._create_appsync_endpoint(function=functions["AppsyncResolverHandler"])
 
     def _create_alb(self, function: Function):
         vpc = ec2.Vpc.from_lookup(
@@ -84,3 +87,40 @@ def _create_lambda_function_url(self, function: Function):
         # Maintenance: move auth to IAM when we create sigv4 builders
         function_url = function.add_function_url(auth_type=FunctionUrlAuthType.AWS_IAM)
         CfnOutput(self.stack, "LambdaFunctionUrl", value=function_url.url)
+
+    def _create_appsync_endpoint(self, function: Function):
+        api = appsync.GraphqlApi(
+            self.stack,
+            "Api",
+            name="e2e-tests",
+            schema=appsync.SchemaFile.from_asset(str(Path(self.feature_path, "files/schema.graphql"))),
+            authorization_config=appsync.AuthorizationConfig(
+                default_authorization=appsync.AuthorizationMode(
+                    authorization_type=appsync.AuthorizationType.API_KEY,
+                    api_key_config=appsync.ApiKeyConfig(
+                        description="public key for getting data",
+                        expires=Expiration.after(Duration.hours(25)),
+                        name="API Token",
+                    ),
+                )
+            ),
+            xray_enabled=False,
+        )
+        lambda_datasource = api.add_lambda_data_source("DataSource", lambda_function=function)
+
+        lambda_datasource.create_resolver(
+            "QueryGetAllPostsResolver",
+            type_name="Query",
+            field_name="allPosts",
+        )
+        lambda_datasource.create_resolver(
+            "QueryGetPostResolver",
+            type_name="Query",
+            field_name="getPost",
+        )
+        lambda_datasource.create_resolver(
+            "QueryGetPostRelatedResolver", type_name="Post", field_name="relatedPosts", max_batch_size=10
+        )
+
+        CfnOutput(self.stack, "GraphQLHTTPUrl", value=api.graphql_url)
+        CfnOutput(self.stack, "GraphQLAPIKey", value=api.api_key)

tests/e2e/event_handler/test_appsync_resolvers.py

@@ -0,0 +1,108 @@
+import json
+
+import pytest
+from requests import Request
+
+from tests.e2e.utils import data_fetcher
+
+
+@pytest.fixture
+def appsync_endpoint(infrastructure: dict) -> str:
+    return infrastructure["GraphQLHTTPUrl"]
+
+
+@pytest.fixture
+def appsync_access_key(infrastructure: dict) -> str:
+    return infrastructure["GraphQLAPIKey"]
+
+
+@pytest.mark.xdist_group(name="event_handler")
+def test_appsync_get_all_posts(appsync_endpoint, appsync_access_key):
+    # GIVEN
+    body = {
+        "query": "query MyQuery { allPosts { post_id }}",
+        "variables": None,
+        "operationName": "MyQuery",
+    }
+
+    # WHEN
+    response = data_fetcher.get_http_response(
+        Request(
+            method="POST",
+            url=appsync_endpoint,
+            json=body,
+            headers={"x-api-key": appsync_access_key, "Content-Type": "application/json"},
+        )
+    )
+
+    # THEN expect a HTTP 200 response and content return list of Posts
+    assert response.status_code == 200
+    assert response.content is not None
+
+    data = json.loads(response.content.decode("ascii"))["data"]
+
+    assert data["allPosts"] is not None
+    assert len(data["allPosts"]) > 0
+
+
+@pytest.mark.xdist_group(name="event_handler")
+def test_appsync_get_post(appsync_endpoint, appsync_access_key):
+    # GIVEN
+    post_id = "1"
+    body = {
+        "query": f'query MyQuery {{ getPost(post_id: "{post_id}") {{ post_id }} }}',
+        "variables": None,
+        "operationName": "MyQuery",
+    }
+
+    # WHEN
+    response = data_fetcher.get_http_response(
+        Request(
+            method="POST",
+            url=appsync_endpoint,
+            json=body,
+            headers={"x-api-key": appsync_access_key, "Content-Type": "application/json"},
+        )
+    )
+
+    # THEN expect a HTTP 200 response and content return Post id
+    assert response.status_code == 200
+    assert response.content is not None
+
+    data = json.loads(response.content.decode("ascii"))["data"]
+
+    assert data["getPost"]["post_id"] == post_id
+
+
+@pytest.mark.xdist_group(name="event_handler")
+def test_appsync_get_related_posts_batch(appsync_endpoint, appsync_access_key):
+    # GIVEN
+    post_id = "2"
+    related_posts_ids = ["3", "5"]
+
+    body = {
+        "query": f'query MyQuery {{ getPost(post_id: "{post_id}") {{ post_id relatedPosts {{ post_id }} }} }}',
+        "variables": None,
+        "operationName": "MyQuery",
+    }
+
+    # WHEN
+    response = data_fetcher.get_http_response(
+        Request(
+            method="POST",
+            url=appsync_endpoint,
+            json=body,
+            headers={"x-api-key": appsync_access_key, "Content-Type": "application/json"},
+        )
+    )
+
+    # THEN expect a HTTP 200 response and content return Post id with dependent Posts id's
+    assert response.status_code == 200
+    assert response.content is not None
+
+    data = json.loads(response.content.decode("ascii"))["data"]
+
+    assert data["getPost"]["post_id"] == post_id
+    assert len(data["getPost"]["relatedPosts"]) == len(related_posts_ids)
+    for post in data["getPost"]["relatedPosts"]:
+        assert post["post_id"] in related_posts_ids