feat(event-handler): add single resolver functionality for AppSync GraphQL API

[arnabrahman]

Summary

This PR will add the ability to register single resolvers for AppSync GraphQL API.

Changes

• Add new appsync-graphql utility inside event-handler
• For this PR, we are only adding the single resolver functionality for AppSync GraphQL API
• Followed this suggestion and same coding structure for appsync-events
• TypeScript does not natively support unpacking an object as named arguments like Python’s kwargs. We must match the function signature accordingly. So this is not possible, I resolved the arguments as an object.
• I created onQuery and onMutation methods for a single resolver. Although this probably improves the developer experience (DX), I feel like it might not be worth it. It could be merged into a single onResolve method instead. This way, we can avoid another Map lookup. Need opinion on this.

Example usage of the utility: Followed this

import { AppSyncGraphQLResolver } from '@aws-lambda-powertools/event-handler/appsync-graphql';
import { Logger } from '@aws-lambda-powertools/logger';
import type { Context } from 'aws-lambda';

const logger = new Logger();
const app = new AppSyncGraphQLResolver({
  logger: logger
});

const posts: Record<number, Record<string, unknown>> = {
  1: { id: '1', title: 'First book', author: 'Author1', url: 'https://amazon.com/', content: 'SAMPLE TEXT AUTHOR 1 SAMPLE TEXT AUTHOR 1 SAMPLE TEXT AUTHOR 1 SAMPLE TEXT AUTHOR 1 SAMPLE TEXT AUTHOR 1 SAMPLE TEXT AUTHOR 1', ups: '100', downs: '10', },
  2: { id: '2', title: 'Second book', author: 'Author2', url: 'https://amazon.com', content: 'SAMPLE TEXT AUTHOR 2 SAMPLE TEXT AUTHOR 2 SAMPLE TEXT', ups: '100', downs: '10', },
  3: { id: '3', title: 'Third book', author: 'Author3', url: null, content: null, ups: null, downs: null },
  4: { id: '4', title: 'Fourth book', author: 'Author4', url: 'https://www.amazon.com/', content: 'SAMPLE TEXT AUTHOR 4 SAMPLE TEXT AUTHOR 4 SAMPLE TEXT AUTHOR 4 SAMPLE TEXT AUTHOR 4 SAMPLE TEXT AUTHOR 4 SAMPLE TEXT AUTHOR 4 SAMPLE TEXT AUTHOR 4 SAMPLE TEXT AUTHOR 4', ups: '1000', downs: '0', },
  5: { id: '5', title: 'Fifth book', author: 'Author5', url: 'https://www.amazon.com/', content: 'SAMPLE TEXT AUTHOR 5 SAMPLE TEXT AUTHOR 5 SAMPLE TEXT AUTHOR 5 SAMPLE TEXT AUTHOR 5 SAMPLE TEXT', ups: '50', downs: '0', },
};


class Lambda {
  @app.onQuery('getPost')
  async getPost({ id }: { id: number; }) {
    return posts[id];

  }
  @app.onMutation('addPost')
  async addPost({ id, title, author, url, content }: { id:number, title: string, author: string, url: string, content: string, }) {
    posts[id] = { id: String(id), title, author, url, content };
    return posts[id];
  }


  async handler(event: unknown, context: Context) {
    return app.resolve(event, context);
  }
}

const lambda = new Lambda();
export const handler = lambda.handler.bind(lambda);

type Post {
	id: ID!
	author: String!
	title: String
	content: String
	url: String
	ups: Int
	downs: Int
	relatedPosts: [Post]
}

type Mutation {
	addPost(
		id: ID!,
		author: String!,
		title: String,
		content: String,
		url: String
	): Post!
}

type Query {
	getPost(id: ID!): Post
	allPosts: [Post]
}

schema {
	query: Query
	mutation: Mutation
}

Issue number: #1166

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

Disclaimer: We value your time and bandwidth. As such, any pull requests created on non-triaged issues might not be successful.

[arnabrahman]

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

[dreamorosi]

Hi @arnabrahman, thank you so much for the PR.

I'm going to need a bit more time to review this, I was out of office yesterday and catching up with things today.

I expect to provide a first review tomorrow morning.

[leandrodamascena]

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

[arnabrahman]

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

[svozza]

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)
    );
  });

[sonarqubecloud[bot]]

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