I have also begun exploring the implementation of the batch resolver. I am trying the powertools-python batch resolver example.

Schema:

type Post {
	post_id: ID!
	title: String
	author: String
	relatedPosts: [Post]
}

type Query {
	getPost(post_id: ID!): Post
}

schema {
	query: Query
}

from __future__ import annotations

from typing import Any, TypedDict

from aws_lambda_powertools import Logger
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()
logger = Logger()

posts = {
    "1": { "post_id": "1", "title": "First book", "author": "Author1", },
    "2": { "post_id": "2", "title": "Second book", "author": "Author2", },
    "3": { "post_id": "3", "title": "Third book", "author": "Author3", },
    "4": { "post_id": "4", "title": "Fourth book", "author": "Author4", },
    "5": { "post_id": "5", "title": "Fifth book", "author": "Author5", }
}

posts_related = {
    "1": [posts["4"]],
    "2": [posts["3"], posts["5"]],
    "3": [posts["2"], posts["1"]],
    "4": [posts["2"], posts["1"]],
    "5": [],
}


def search_batch_posts(posts: list) -> dict[str, Any]:
    return {post_id: posts_related.get(post_id) for post_id in posts}

class Post(TypedDict, total=False):
    post_id: str
    title: str
    author: str

@app.resolver(type_name="Query", field_name="getPost")
def get_post(
    post_id: str = ""
) -> Post:
    return posts.get(post_id, {})


@app.batch_resolver(type_name="Query", field_name="relatedPosts")
def related_posts(event: list[AppSyncResolverEvent]) -> list[Any]:  
    # Extract all post_ids in order
    post_ids: list = [record.source.get("post_id") for record in event]  

    # Get unique post_ids while preserving order
    unique_post_ids = list(dict.fromkeys(post_ids))
    # Fetch posts in a single batch operation
    fetched_posts = search_batch_posts(unique_post_ids)

    # Return results in original order
    return [fetched_posts.get(post_id) for post_id in post_ids]


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

If I try this query

query MyQuery {
  getPost(post_id: "2") {
    relatedPosts {
      post_id
      author
      relatedPosts {
        post_id
        author
      }
    }
  }
}

There is an error: [ERROR] ResolverNotFoundError: No resolver found for 'Post.relatedPosts' Traceback (most recent call last):

If I change the typeName to Post, it works -> @app.batch_resolver(type_name="Post", field_name="relatedPosts")

Result:

{
  "data": {
    "getPost": {
      "relatedPosts": [
        {
          "post_id": "3",
          "author": "Author3",
          "relatedPosts": [
            {
              "post_id": "2",
              "author": "Author2"
            },
            {
              "post_id": "1",
              "author": "Author1"
            }
          ]
        },
        {
          "post_id": "5",
          "author": "Author5",
          "relatedPosts": []
        }
      ]
    }
  }
}

The reason is probably because parentTypeName value is Post.

Not sure what I am doing wrong here. @dreamorosi @leandrodamascena

Thanks for finding a bug in the Python documentation, @arnabrahman! You always exceed our standards with these catches ❤️ ! Yeah, the valid value is Post and not Query because when resolving a batch schema the initial Query (getPost) make a Post to relatedPosts. Tbh this should be Query as well, but in this case I think we need to pass the parent post as parameter and would difficult the example.

Thanks

@dreamorosi Thanks for taking the time to review this. I will look at the suggestions and get back on this.

This is great work. One question I have: both the AppSync events and Bedrock agents event handlers pass the event and the context objects into their resolvers and I think we should do the same here. Here's an example of what I mean:

it('tool function has access to the event variable', async () => {
    // Prepare
    const app = new BedrockAgentFunctionResolver();

    app.tool(
      async (_params, options) => {
        return options?.event;
      },
      {
        name: 'event-accessor',
        description: 'Accesses the event object',
      }
    );

    const event = createEvent('event-accessor');

    // Act
    const result = await app.resolve(event, context);

    // Assess
    expect(result.response.function).toEqual('event-accessor');
    expect(result.response.functionResponse.responseBody.TEXT.body).toEqual(
      JSON.stringify(event)
    );
  });

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud