Thanks for opening your first issue here! We'll come back to you as soon as we can.

hey @cponfick-bhs thank you for raising this, it makes sense. We'd need to do some digging and experiment with a few UX - would you be able to share a sample event with batch resolver?

That will help speed this up as we'd need to create some tests, find most optimal type definitions, check whether we need some additional fields/methods, and then refactor in a non-breaking change fashion.

Thanks a lot!

hey @cponfick-bhs thank you for raising this, it makes sense. We'd need to do some digging and experiment with a few UX - would you be able to share a sample event with batch resolver?

Sure, I can construct an example.
Assume the following GraphQl-Schema:

type Query{
   getProducts(): [Product]
}

type Product {
   name: String
   comments: [Comment!]
}

type Comment {
   text: String!
   userName: String!
}

The resolvers are attached as follows:

• ParentTypeName: Query and FieldName: getProducts
• ParentTypeName: Product and FieldName: comments

When using the usual Invoke, Appsync creates one event for the Query:

{
       "arguments": {},
       "identity": {
           "claims": {},
           "defaultAuthStrategy": "ALLOW",
           "groups": null,
           "issuer": "",
           "sourceIp": [""],
           "sub": "",
           "username": ""
       },
       "source": null,
       "request": {
           "headers": {},
           "domainName": null
       },
       "prev": null,
       "info": {
           "selectionSetList":["EVERYTHING"],
           "fieldName": "getProducts",
           "parentTypeName": "Query",
           "variables": {}
       },
       "stash": {}
   }

and for every product, an additional event and lambda invocation for the comments:

{
        "arguments": {},
        "identity": {
            "claims": {},
            "defaultAuthStrategy": "ALLOW",
            "groups": null,
            "issuer": "",
            "sourceIp": [""],
            "sub": "",
            "username": ""
        },
        "source": {THE FIELDS OF THE PRODUCT},
        "request": {
            "headers": {},
            "domainName": null
        },
        "prev": null,
        "info": {
            "selectionSetList":["EVERYTHING"],
            "fieldName": "comments",
            "parentTypeName": "Product",
            "variables": {}
        },
        "stash": {}
    }

Hence, for every Product an additional lambda invoke is executed.

When using "BatchInvoke", the first event remains the same, but the calls that return the comments for each product are "merged" into one lambda call with a list of events.

[{
        "arguments": {},
        "identity": {
            "claims": {},
            "defaultAuthStrategy": "ALLOW",
            "groups": null,
            "issuer": "",
            "sourceIp": [""],
            "sub": "",
            "username": ""
        },
        "source": {THE FIELDS OF THE PRODUCT 0},
        "request": {
            "headers": {},
            "domainName": null
        },
        "prev": null,
        "info": {
            "selectionSetList":["EVERYTHING"],
            "fieldName": "comments",
            "parentTypeName": "Product",
            "variables": {}
        },
        "stash": {}
    },
{
        "arguments": {},
        "identity": {
            "claims": {},
            "defaultAuthStrategy": "ALLOW",
            "groups": null,
            "issuer": "",
            "sourceIp": [""],
            "sub": "",
            "username": ""
        },
        "source": {THE FIELDS OF THE PRODUCT 1},
        "request": {
            "headers": {},
            "domainName": null
        },
        "prev": null,
        "info": {
            "selectionSetList":["EVERYTHING"],
            "fieldName": "comments",
            "parentTypeName": "Product",
            "variables": {}
        },
        "stash": {}
    },
{
        "arguments": {},
        "identity": {
            "claims": {},
            "defaultAuthStrategy": "ALLOW",
            "groups": null,
            "issuer": "",
            "sourceIp": [""],
            "sub": "",
            "username": ""
        },
        "source": {THE FIELDS OF THE PRODUCT 2},
        "request": {
            "headers": {},
            "domainName": null
        },
        "prev": null,
        "info": {
            "selectionSetList":["EVERYTHING"],
            "fieldName": "comments",
            "parentTypeName": "Product",
            "variables": {}
        },
        "stash": {}
    }]

Appsync now expects a list of responses associated with the corresponding sources. The fields of the events depend on the ResponseMappingTemplate used for the resolver.

@heitorlessa I hope this helps a bit.

Work Around

Currently, I work around this issue using the following AppSyncResolverEvent:

class MyAppSyncResolverEvent(appsync.AppSyncResolverEvent):
    def __init__(self, data: Union[dict, list]):
        if isinstance(data, dict):
            super().__init__(data)
        else:
            super().__init__(data[0])
            self._sources = [event.get('source', {}) for event in data]

    @property
    def arguments(self) -> Dict[str, Any]:
        if isinstance(self._data, dict):
            return self["arguments"]
        else:
            return {}

    @property
    def sources(self) -> List[Dict[str, Any]]:
        return self._sources

I must admit that this is not clean, but for now it works.

Updating the status so we can revisit early next year.

+1
We had to build to a similar class as created by @cponfick-bhs

Hi @cponfick-bhs and @thenamanpatwari!

I was on parental leave and I'm returning to work now. I created an example similar to the one reported here to understand the behavior and look for the best solution.

By the middle of next week I'll be back on this issue with some updates on what I found.

Thanks

Hi all! Updating here, I still haven't been able to fully analyze this and a possible solution. I'm focused on bringing you an update ASAP.

Leandro and @mploski will be updating it this week

I was able to replicate the issue. Indeed, if we enable batch processing on AppSync side, resolver lambda gets list of events which is something we don't expect. Execution fails with [ERROR] AttributeError: 'list' object has no attribute 'get' in aws_lambda_powertools/utilities/data_classes/appsync_resolver_event.py. Attached SAM project zip with lambda and template that allow us to replicate it reliably. appsync-batch-debug.tar.gz
I will work together with @leandrodamascena on the solution.

UPDATE: @mploski is working on an update by EOW.

UPDATE: @mploski is working on an update by EOW.

Hey sorry for lack of response here before. I was fully booked with other things for last month. We agreed that @leandrodamascena will support me in the further development so we can close it soon

Hey everyone! We are working on this PR #1998 to add this support as quickly as possible. The most up-to-date situation is:

We will have an internal meeting with the AppSync Service team to discuss error handling when consuming an AppSync batch process. We want to hear from them about use cases and what challenges customers are facing.

After this meeting, I will update this PR. The expectation of having this feature in the next release remains, but I will update this PR if we have any additional news from this meeting.

Thanks

Update. We're going through the final documentation touches to get this released in the next version.

Since last week, we've added partial failure support so parts of the batch item could fail, letting AppSync or your own response mapping templates propagate partial errors accordingly.

from typing import Dict, Optional

from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
from aws_lambda_powertools.utilities.typing import LambdaContext

app = AppSyncResolver()


posts_related = {
    "1": {"title": "post1"},
    "2": {"title": "post2"},
    "3": {"title": "post3"},
}


class PostRelatedNotFound(Exception):
    ...


@app.batch_resolver(type_name="Query", field_name="relatedPosts", raise_on_error=True)  
def related_posts(event: AppSyncResolverEvent, post_id: str) -> Optional[Dict]:
    post_found = posts_related.get(post_id, None)

    if not post_found:
        raise PostRelatedNotFound(f"Unable to find a related post with ID {post_id}.")

    return post_found


def lambda_handler(event, context: LambdaContext) -> dict:
    return app.resolve(event, context)

We're making a patch release today with numerous bugfixes, and the next feature release will have this feature released.

Deepest apologies on the delay here