Skip to content

GraphQL API

Event handler for AWS AppSync Direct Lambda Resolver and Amplify GraphQL Transformer.

Key Features

  • Automatically parse API arguments to function arguments
  • Choose between strictly match a GraphQL field name or all of them to a function
  • Integrates with Data classes utilities to access resolver and identity information
  • Works with both Direct Lambda Resolver and Amplify GraphQL Transformer @function directive
  • Support async Python 3.8+ functions, and generators

Terminology

Direct Lambda Resolver. A custom AppSync Resolver to bypass the use of Apache Velocity Template (VTL) and automatically map your function's response to a GraphQL field.

Amplify GraphQL Transformer. Custom GraphQL directives to define your application's data model using Schema Definition Language (SDL). Amplify CLI uses these directives to convert GraphQL SDL into full descriptive AWS CloudFormation templates.

Getting started

Required resources

You must have an existing AppSync GraphQL API and IAM permissions to invoke your Lambda function. That said, there is no additional permissions to use this utility.

This is the sample infrastructure we are using for the initial examples with a AppSync Direct Lambda Resolver.

Tip: Designing GraphQL Schemas for the first time?

Visit AWS AppSync schema documentation for understanding how to define types, nesting, and pagination.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
schema {
    query: Query
}

type Query {
    getTodo(id: ID!): Todo
    listTodos: [Todo]
}

type Todo {
    id: ID!
    userId: String
    title: String
    completed: Boolean
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Description: Hello world Direct Lambda Resolver

Globals:
  Function:
    Timeout: 5
    Runtime: python3.9
    Tracing: Active
    Environment:
      Variables:
        # Powertools env vars: https://awslabs.github.io/aws-lambda-powertools-python/latest/#environment-variables
        LOG_LEVEL: INFO
        POWERTOOLS_LOGGER_SAMPLE_RATE: 0.1
        POWERTOOLS_LOGGER_LOG_EVENT: true
        POWERTOOLS_SERVICE_NAME: example

Resources:
  TodosFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: getting_started_graphql_api_resolver.lambda_handler
      CodeUri: ../src
      Description: Sample Direct Lambda Resolver

  # IAM Permissions and Roles

  AppSyncServiceRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              Service:
                - "appsync.amazonaws.com"
            Action:
              - "sts:AssumeRole"

  InvokeLambdaResolverPolicy:
    Type: "AWS::IAM::Policy"
    Properties:
      PolicyName: "DirectAppSyncLambda"
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Action: "lambda:invokeFunction"
            Resource:
              - !GetAtt TodosFunction.Arn
      Roles:
        - !Ref AppSyncServiceRole

  # GraphQL API

  TodosApi:
    Type: "AWS::AppSync::GraphQLApi"
    Properties:
      Name: TodosApi
      AuthenticationType: "API_KEY"
      XrayEnabled: true

  TodosApiKey:
    Type: AWS::AppSync::ApiKey
    Properties:
      ApiId: !GetAtt TodosApi.ApiId

  TodosApiSchema:
    Type: "AWS::AppSync::GraphQLSchema"
    Properties:
      ApiId: !GetAtt TodosApi.ApiId
      Definition: |
        schema {
            query:Query
        }

        type Query {
            getTodo(id: ID!): Todo
            listTodos: [Todo]
        }

        type Todo {
            id: ID!
            userId: String
            title: String
            completed: Boolean
        }

  # Lambda Direct Data Source and Resolver

  TodosFunctionDataSource:
    Type: "AWS::AppSync::DataSource"
    Properties:
      ApiId: !GetAtt TodosApi.ApiId
      Name: "HelloWorldLambdaDirectResolver"
      Type: "AWS_LAMBDA"
      ServiceRoleArn: !GetAtt AppSyncServiceRole.Arn
      LambdaConfig:
        LambdaFunctionArn: !GetAtt TodosFunction.Arn

  ListTodosResolver:
    Type: "AWS::AppSync::Resolver"
    Properties:
      ApiId: !GetAtt TodosApi.ApiId
      TypeName: "Query"
      FieldName: "listTodos"
      DataSourceName: !GetAtt TodosFunctionDataSource.Name

  GetTodoResolver:
    Type: "AWS::AppSync::Resolver"
    Properties:
      ApiId: !GetAtt TodosApi.ApiId
      TypeName: "Query"
      FieldName: "getTodo"
      DataSourceName: !GetAtt TodosFunctionDataSource.Name

Outputs:
  TodosFunction:
    Description: "Hello World Lambda Function ARN"
    Value: !GetAtt TodosFunction.Arn

  TodosApi:
    Value: !GetAtt TodosApi.Arn

Resolver decorator

You can define your functions to match GraphQL types and fields with the app.resolver() decorator.

Here's an example where we have two separate functions to resolve getTodo and listTodos fields within the Query type. For completion, we use Scalar type utilities to generate the right output based on our schema definition.

Important

GraphQL arguments are passed as function keyword arguments.

Example

The GraphQL Query getTodo(id: "todo_id_value") will call get_todo as get_todo(id="todo_id_value").

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import sys

if sys.version_info >= (3, 8):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

from typing import List

import requests
from requests import Response

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = AppSyncResolver()


class Todo(TypedDict, total=False):
    id: str  # noqa AA03 VNE003, required due to GraphQL Schema
    userId: str
    title: str
    completed: bool


@app.resolver(type_name="Query", field_name="getTodo")
@tracer.capture_method
def get_todo(
    id: str = "",  # noqa AA03 VNE003 shadows built-in id to match query argument, e.g., getTodo(id: "some_id")
) -> Todo:
    logger.info(f"Fetching Todo {id}")
    todos: Response = requests.get(f"https://jsonplaceholder.typicode.com/todos/{id}")
    todos.raise_for_status()

    return todos.json()


@app.resolver(type_name="Query", field_name="listTodos")
@tracer.capture_method
def list_todos() -> List[Todo]:
    todos: Response = requests.get("https://jsonplaceholder.typicode.com/todos")
    todos.raise_for_status()

    # for brevity, we'll limit to the first 10 only
    return todos.json()[:10]


@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
schema {
    query: Query
}

type Query {
    getTodo(id: ID!): Todo
    listTodos: [Todo]
}

type Todo {
    id: ID!
    userId: String
    title: String
    completed: Boolean
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
{
    "arguments": {
        "id": "7e362732-c8cd-4405-b090-144ac9b38960"
    },
    "identity": null,
    "source": null,
    "request": {
        "headers": {
            "x-forwarded-for": "1.2.3.4, 5.6.7.8",
            "accept-encoding": "gzip, deflate, br",
            "cloudfront-viewer-country": "NL",
            "cloudfront-is-tablet-viewer": "false",
            "referer": "https://eu-west-1.console.aws.amazon.com/appsync/home?region=eu-west-1",
            "via": "2.0 9fce949f3749407c8e6a75087e168b47.cloudfront.net (CloudFront)",
            "cloudfront-forwarded-proto": "https",
            "origin": "https://eu-west-1.console.aws.amazon.com",
            "x-api-key": "da1-c33ullkbkze3jg5hf5ddgcs4fq",
            "content-type": "application/json",
            "x-amzn-trace-id": "Root=1-606eb2f2-1babc433453a332c43fb4494",
            "x-amz-cf-id": "SJw16ZOPuMZMINx5Xcxa9pB84oMPSGCzNOfrbJLvd80sPa0waCXzYQ==",
            "content-length": "114",
            "x-amz-user-agent": "AWS-Console-AppSync/",
            "x-forwarded-proto": "https",
            "host": "ldcvmkdnd5az3lm3gnf5ixvcyy.appsync-api.eu-west-1.amazonaws.com",
            "accept-language": "en-US,en;q=0.5",
            "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:78.0) Gecko/20100101 Firefox/78.0",
            "cloudfront-is-desktop-viewer": "true",
            "cloudfront-is-mobile-viewer": "false",
            "accept": "*/*",
            "x-forwarded-port": "443",
            "cloudfront-is-smarttv-viewer": "false"
        }
    },
    "prev": null,
    "info": {
        "parentTypeName": "Query",
        "selectionSetList": [
            "title",
            "id"
        ],
        "selectionSetGraphQL": "{\n  title\n  id\n}",
        "fieldName": "getTodo",
        "variables": {}
    },
    "stash": {}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
{
    "arguments": {},
    "identity": null,
    "source": null,
    "request": {
        "headers": {
            "x-forwarded-for": "1.2.3.4, 5.6.7.8",
            "accept-encoding": "gzip, deflate, br",
            "cloudfront-viewer-country": "NL",
            "cloudfront-is-tablet-viewer": "false",
            "referer": "https://eu-west-1.console.aws.amazon.com/appsync/home?region=eu-west-1",
            "via": "2.0 9fce949f3749407c8e6a75087e168b47.cloudfront.net (CloudFront)",
            "cloudfront-forwarded-proto": "https",
            "origin": "https://eu-west-1.console.aws.amazon.com",
            "x-api-key": "da1-c33ullkbkze3jg5hf5ddgcs4fq",
            "content-type": "application/json",
            "x-amzn-trace-id": "Root=1-606eb2f2-1babc433453a332c43fb4494",
            "x-amz-cf-id": "SJw16ZOPuMZMINx5Xcxa9pB84oMPSGCzNOfrbJLvd80sPa0waCXzYQ==",
            "content-length": "114",
            "x-amz-user-agent": "AWS-Console-AppSync/",
            "x-forwarded-proto": "https",
            "host": "ldcvmkdnd5az3lm3gnf5ixvcyy.appsync-api.eu-west-1.amazonaws.com",
            "accept-language": "en-US,en;q=0.5",
            "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:78.0) Gecko/20100101 Firefox/78.0",
            "cloudfront-is-desktop-viewer": "true",
            "cloudfront-is-mobile-viewer": "false",
            "accept": "*/*",
            "x-forwarded-port": "443",
            "cloudfront-is-smarttv-viewer": "false"
        }
    },
    "prev": null,
    "info": {
        "parentTypeName": "Query",
        "selectionSetList": [
            "id",
            "title"
        ],
        "selectionSetGraphQL": "{\n  id\n  title\n}",
        "fieldName": "listTodos",
        "variables": {}
    },
    "stash": {}
}

Scalar functions

When working with AWS AppSync Scalar types, you might want to generate the same values for data validation purposes.

For convenience, the most commonly used values are available as functions within scalar_types_utils module.

Creating key scalar values with scalar_types_utils
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from aws_lambda_powertools.utilities.data_classes.appsync.scalar_types_utils import (
    aws_date,
    aws_datetime,
    aws_time,
    aws_timestamp,
    make_id,
)

# Scalars: https://docs.aws.amazon.com/appsync/latest/devguide/scalars.html

my_id: str = make_id()  # Scalar: ID!
my_date: str = aws_date()  # Scalar: AWSDate
my_timestamp: str = aws_time()  # Scalar: AWSTime
my_datetime: str = aws_datetime()  # Scalar: AWSDateTime
my_epoch_timestamp: int = aws_timestamp()  # Scalar: AWSTimestamp

Here's a table with their related scalar as a quick reference:

Scalar type Scalar function Sample value
ID scalar_types_utils.make_id e916c84d-48b6-484c-bef3-cee3e4d86ebf
AWSDate scalar_types_utils.aws_date 2022-07-08Z
AWSTime scalar_types_utils.aws_time 15:11:00.189Z
AWSDateTime scalar_types_utils.aws_datetime 2022-07-08T15:11:00.189Z
AWSTimestamp scalar_types_utils.aws_timestamp 1657293060

Advanced

Nested mappings

Note

The following examples use a more advanced schema. These schemas differ from initial sample infrastructure we used earlier.

You can nest app.resolver() decorator multiple times when resolving fields with the same return value.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import sys

if sys.version_info >= (3, 8):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

from typing import List

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = AppSyncResolver()


class Location(TypedDict, total=False):
    id: str  # noqa AA03 VNE003, required due to GraphQL Schema
    name: str
    description: str
    address: str


@app.resolver(field_name="listLocations")
@app.resolver(field_name="locations")
@tracer.capture_method
def get_locations(name: str, description: str = "") -> List[Location]:  # match GraphQL Query arguments
    return [{"name": name, "description": description}]


@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
schema {
    query: Query
}

type Query {
    listLocations: [Location]
}

type Location {
    id: ID!
    name: String!
    description: String
    address: String
}

type Merchant {
    id: String!
    name: String!
    description: String
    locations: [Location]
}

Async functions

For Lambda Python3.8+ runtime, this utility supports async functions when you use in conjunction with asyncio.run.

Resolving GraphQL resolvers async
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import asyncio
import sys

if sys.version_info >= (3, 8):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

from typing import List

import aiohttp

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.tracing import aiohttp_trace_config
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = AppSyncResolver()


class Todo(TypedDict, total=False):
    id: str  # noqa AA03 VNE003, required due to GraphQL Schema
    userId: str
    title: str
    completed: bool


@app.resolver(type_name="Query", field_name="listTodos")
async def list_todos() -> List[Todo]:
    async with aiohttp.ClientSession(trace_configs=[aiohttp_trace_config()]) as session:
        async with session.get("https://jsonplaceholder.typicode.com/todos") as resp:
            return await resp.json()


@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    result = app.resolve(event, context)

    return asyncio.run(result)

Amplify GraphQL Transformer

Assuming you have Amplify CLI installed, create a new API using amplify add api and use the following GraphQL Schema.

Example GraphQL Schema
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
@model
type Merchant {
    id: String!
    name: String!
    description: String
    # Resolves to `common_field`
    commonField: String  @function(name: "merchantInfo-${env}")
}

type Location {
    id: ID!
    name: String!
    address: String
    # Resolves to `common_field`
    commonField: String  @function(name: "merchantInfo-${env}")
}

type Query {
  # List of locations resolves to `list_locations`
  listLocations(page: Int, size: Int): [Location] @function(name: "merchantInfo-${env}")
  # List of locations resolves to `list_locations`
  findMerchant(search: str): [Merchant] @function(name: "searchMerchant-${env}")
}

Create two new basic Python functions via amplify add function.

Note

Amplify CLI generated functions use Pipenv as a dependency manager. Your function source code is located at amplify/backend/function/your-function-name.

Within your function's folder, add Powertools as a dependency with pipenv install aws-lambda-powertools.

Use the following code for merchantInfo and searchMerchant functions respectively.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import sys

if sys.version_info >= (3, 8):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

from typing import List

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.data_classes.appsync import scalar_types_utils
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = AppSyncResolver()


class Location(TypedDict, total=False):
    id: str  # noqa AA03 VNE003, required due to GraphQL Schema
    name: str
    description: str
    address: str
    commonField: str


@app.resolver(type_name="Query", field_name="listLocations")
def list_locations(page: int = 0, size: int = 10) -> List[Location]:
    return [{"id": scalar_types_utils.make_id(), "name": "Smooth Grooves"}]


@app.resolver(field_name="commonField")
def common_field() -> str:
    # Would match all fieldNames matching 'commonField'
    return scalar_types_utils.make_id()


@tracer.capture_lambda_handler
@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
import sys

if sys.version_info >= (3, 8):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

from typing import List

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.data_classes.appsync import scalar_types_utils
from aws_lambda_powertools.utilities.typing import LambdaContext

app = AppSyncResolver()
tracer = Tracer()
logger = Logger()


class Merchant(TypedDict, total=False):
    id: str  # noqa AA03 VNE003, required due to GraphQL Schema
    name: str
    description: str
    commonField: str


@app.resolver(type_name="Query", field_name="findMerchant")
def find_merchant(search: str) -> List[Merchant]:
    merchants: List[Merchant] = [
        {
            "id": scalar_types_utils.make_id(),
            "name": "Parry-Wood",
            "description": "Possimus doloremque tempora harum deleniti eum.",
        },
        {
            "id": scalar_types_utils.make_id(),
            "name": "Shaw, Owen and Jones",
            "description": "Aliquam iste architecto suscipit in.",
        },
    ]

    return [merchant for merchant in merchants if search == merchant["name"]]


@tracer.capture_lambda_handler
@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
{
    "typeName": "Query",
    "fieldName": "listLocations",
    "arguments": {
        "page": 2,
        "size": 1
    },
    "identity": {
        "claims": {
            "iat": 1615366261
        },
        "username": "treid"
    },
    "request": {
        "headers": {
            "x-amzn-trace-id": "Root=1-60488877-0b0c4e6727ab2a1c545babd0",
            "x-forwarded-for": "127.0.0.1",
            "cloudfront-viewer-country": "NL",
            "x-api-key": "da1-c33ullkbkze3jg5hf5ddgcs4fq"
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
{
    "typeName": "Merchant",
    "fieldName": "commonField",
    "arguments": {},
    "identity": {
        "claims": {
            "iat": 1615366261
        },
        "username": "marieellis"
    },
    "request": {
        "headers": {
            "x-amzn-trace-id": "Root=1-60488877-0b0c4e6727ab2a1c545babd0",
            "x-forwarded-for": "127.0.0.1"
        }
    },
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
{
    "typeName": "Query",
    "fieldName": "findMerchant",
    "arguments": {
        "search": "Parry-Wood"
    },
    "identity": {
        "claims": {
            "iat": 1615366261
        },
        "username": "wwilliams"
    },
    "request": {
        "headers": {
            "x-amzn-trace-id": "Root=1-60488877-0b0c4e6727ab2a1c545babd0",
            "x-forwarded-for": "127.0.0.1"
        }
    },
}

Custom data models

You can subclass AppSyncResolverEvent to bring your own set of methods to handle incoming events, by using data_model param in the resolve method.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
import sys

if sys.version_info >= (3, 8):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

from typing import List

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.data_classes.appsync import scalar_types_utils
from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import (
    AppSyncResolverEvent,
)
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = AppSyncResolver()


class Location(TypedDict, total=False):
    id: str  # noqa AA03 VNE003, required due to GraphQL Schema
    name: str
    description: str
    address: str
    commonField: str


class MyCustomModel(AppSyncResolverEvent):
    @property
    def country_viewer(self) -> str:
        return self.get_header_value(name="cloudfront-viewer-country", default_value="", case_sensitive=False)  # type: ignore[return-value] # sentinel typing # noqa: E501

    @property
    def api_key(self) -> str:
        return self.get_header_value(name="x-api-key", default_value="", case_sensitive=False)  # type: ignore[return-value] # sentinel typing # noqa: E501


@app.resolver(type_name="Query", field_name="listLocations")
def list_locations(page: int = 0, size: int = 10) -> List[Location]:
    # additional properties/methods will now be available under current_event
    logger.debug(f"Request country origin: {app.current_event.country_viewer}")  # type: ignore[attr-defined]
    return [{"id": scalar_types_utils.make_id(), "name": "Perry, James and Carroll"}]


@tracer.capture_lambda_handler
@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context, data_model=MyCustomModel)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
schema {
    query: Query
}

type Query {
    listLocations: [Location]
}

type Location {
    id: ID!
    name: String!
    description: String
    address: String
}

type Merchant {
    id: String!
    name: String!
    description: String
    locations: [Location]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
 {
     "typeName": "Query",
     "fieldName": "listLocations",
     "arguments": {
         "page": 2,
         "size": 1
     },
     "identity": {
         "claims": {
             "iat": 1615366261
         },
         "username": "treid"
     },
     "request": {
         "headers": {
             "x-amzn-trace-id": "Root=1-60488877-0b0c4e6727ab2a1c545babd0",
             "x-forwarded-for": "127.0.0.1",
             "cloudfront-viewer-country": "NL",
             "x-api-key": "da1-c33ullkbkze3jg5hf5ddgcs4fq"
         }
     }
 }

Split operations with Router

Tip

Read the considerations section for trade-offs between monolithic and micro functions, as it's also applicable here.

As you grow the number of related GraphQL operations a given Lambda function should handle, it is natural to split them into separate files to ease maintenance - That's when the Router feature comes handy.

Let's assume you have split_operation.py as your Lambda function entrypoint and routes in split_operation_module.py. This is how you'd use the Router feature.

We import Router instead of AppSyncResolver; syntax wise is exactly the same.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import sys

if sys.version_info >= (3, 8):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

from typing import List

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler.appsync import Router

tracer = Tracer()
logger = Logger()
router = Router()


class Location(TypedDict, total=False):
    id: str  # noqa AA03 VNE003, required due to GraphQL Schema
    name: str
    description: str
    address: str


@router.resolver(field_name="listLocations")
@router.resolver(field_name="locations")
@tracer.capture_method
def get_locations(name: str, description: str = "") -> List[Location]:  # match GraphQL Query arguments
    return [{"name": name, "description": description}]

We use include_router method and include all location operations registered in the router global object.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
import split_operation_module

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = AppSyncResolver()
app.include_router(split_operation_module.router)


@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

Sharing contextual data

You can use append_context when you want to share data between your App and Router instances. Any data you share will be available via the context dictionary available in your App or Router context.

Info

For safety, we always clear any data available in the context dictionary after each invocation.

Tip

This can also be useful for middlewares injecting contextual information before a request is processed.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import split_operation_append_context_module

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = AppSyncResolver()
app.include_router(split_operation_append_context_module.router)


@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    app.append_context(is_admin=True)  # arbitrary number of key=value data
    return app.resolve(event, context)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import sys

if sys.version_info >= (3, 8):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

from typing import List

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler.appsync import Router

tracer = Tracer()
logger = Logger()
router = Router()


class Location(TypedDict, total=False):
    id: str  # noqa AA03 VNE003, required due to GraphQL Schema
    name: str
    description: str
    address: str


@router.resolver(field_name="listLocations")
@router.resolver(field_name="locations")
@tracer.capture_method
def get_locations(name: str, description: str = "") -> List[Location]:  # match GraphQL Query arguments
    is_admin: bool = router.context.get("is_admin", False)
    return [{"name": name, "description": description}] if is_admin else []

Testing your code

You can test your resolvers by passing a mocked or actual AppSync Lambda event that you're expecting.

You can use either app.resolve(event, context) or simply app(event, context).

Here's an example of how you can test your synchronous resolvers:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import json
from dataclasses import dataclass
from pathlib import Path

import pytest
from assert_graphql_response_module import Location, app  # instance of AppSyncResolver


@pytest.fixture
def lambda_context():
    @dataclass
    class LambdaContext:
        function_name: str = "test"
        memory_limit_in_mb: int = 128
        invoked_function_arn: str = "arn:aws:lambda:eu-west-1:123456789012:function:test"
        aws_request_id: str = "da658bd3-2d6f-4e7b-8ec2-937234644fdc"

    return LambdaContext()


def test_direct_resolver(lambda_context):
    # GIVEN
    fake_event = json.loads(Path("assert_graphql_response.json").read_text())

    # WHEN
    result: list[Location] = app(fake_event, lambda_context)

    # THEN
    assert result[0]["name"] == "Perkins-Reed"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import sys

if sys.version_info >= (3, 8):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

from typing import List

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = AppSyncResolver()


class Location(TypedDict, total=False):
    id: str  # noqa AA03 VNE003, required due to GraphQL Schema
    name: str
    description: str
    address: str


@app.resolver(field_name="listLocations")
@app.resolver(field_name="locations")
@tracer.capture_method
def get_locations(name: str, description: str = "") -> List[Location]:  # match GraphQL Query arguments
    return [{"name": name, "description": description}]


@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
{
    "typeName": "Query",
    "fieldName": "listLocations",
    "arguments": {
        "name": "Perkins-Reed",
        "description": "Nulla sed amet. Earum libero qui sunt perspiciatis. Non aliquid accusamus."
    },
    "selectionSetList": [
        "id",
        "name"
    ],
    "identity": {
        "claims": {
            "sub": "192879fc-a240-4bf1-ab5a-d6a00f3063f9",
            "email_verified": true,
            "iss": "https://cognito-idp.us-west-2.amazonaws.com/us-west-xxxxxxxxxxx",
            "phone_number_verified": false,
            "cognito:username": "jdoe",
            "aud": "7471s60os7h0uu77i1tk27sp9n",
            "event_id": "bc334ed8-a938-4474-b644-9547e304e606",
            "token_use": "id",
            "auth_time": 1599154213,
            "phone_number": "+19999999999",
            "exp": 1599157813,
            "iat": 1599154213,
            "email": "jdoe@email.com"
        },
        "defaultAuthStrategy": "ALLOW",
        "groups": null,
        "issuer": "https://cognito-idp.us-west-2.amazonaws.com/us-west-xxxxxxxxxxx",
        "sourceIp": [
            "1.1.1.1"
        ],
        "sub": "192879fc-a240-4bf1-ab5a-d6a00f3063f9",
        "username": "jdoe"
    },
    "request": {
        "headers": {
            "x-amzn-trace-id": "Root=1-60488877-0b0c4e6727ab2a1c545babd0",
            "x-forwarded-for": "127.0.0.1",
            "cloudfront-viewer-country": "NL",
            "x-api-key": "da1-c33ullkbkze3jg5hf5ddgcs4fq"
        }
    }
}

And an example for testing asynchronous resolvers. Note that this requires the pytest-asyncio package. This tests a specific async GraphQL operation.

Note

Alternatively, you can continue call lambda_handler function synchronously as it'd run asyncio.run to await for the coroutine to complete.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import json
from dataclasses import dataclass
from pathlib import Path
from typing import List

import pytest
from assert_async_graphql_response_module import (  # instance of AppSyncResolver
    Todo,
    app,
)


@pytest.fixture
def lambda_context():
    @dataclass
    class LambdaContext:
        function_name: str = "test"
        memory_limit_in_mb: int = 128
        invoked_function_arn: str = "arn:aws:lambda:eu-west-1:123456789012:function:test"
        aws_request_id: str = "da658bd3-2d6f-4e7b-8ec2-937234644fdc"

    return LambdaContext()


@pytest.mark.asyncio
async def test_async_direct_resolver(lambda_context):
    # GIVEN
    fake_event = json.loads(Path("assert_async_graphql_response.json").read_text())

    # WHEN
    result: List[Todo] = await app(fake_event, lambda_context)
    # alternatively, you can also run a sync test against `lambda_handler`
    # since `lambda_handler` awaits the coroutine to complete

    # THEN
    assert result[0]["userId"] == 1
    assert result[0]["id"] == 1
    assert result[0]["completed"] is False

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
import sys

if sys.version_info >= (3, 8):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

import asyncio
from typing import List

import aiohttp

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.tracing import aiohttp_trace_config
from aws_lambda_powertools.utilities.typing import LambdaContext

tracer = Tracer()
logger = Logger()
app = AppSyncResolver()


class Todo(TypedDict, total=False):
    id: str  # noqa AA03 VNE003, required due to GraphQL Schema
    userId: str
    title: str
    completed: bool


@app.resolver(type_name="Query", field_name="listTodos")
async def list_todos() -> List[Todo]:
    async with aiohttp.ClientSession(trace_configs=[aiohttp_trace_config()]) as session:
        async with session.get("https://jsonplaceholder.typicode.com/todos") as resp:
            result: List[Todo] = await resp.json()
            return result[:2]  # first two results to demo assertion


@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    result = app.resolve(event, context)

    return asyncio.run(result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
{
    "typeName": "Query",
    "fieldName": "listTodos",
    "arguments": {},
    "selectionSetList": [
        "id",
        "userId",
        "completed"
    ],
    "identity": {
        "claims": {
            "sub": "192879fc-a240-4bf1-ab5a-d6a00f3063f9",
            "email_verified": true,
            "iss": "https://cognito-idp.us-west-2.amazonaws.com/us-west-xxxxxxxxxxx",
            "phone_number_verified": false,
            "cognito:username": "jdoe",
            "aud": "7471s60os7h0uu77i1tk27sp9n",
            "event_id": "bc334ed8-a938-4474-b644-9547e304e606",
            "token_use": "id",
            "auth_time": 1599154213,
            "phone_number": "+19999999999",
            "exp": 1599157813,
            "iat": 1599154213,
            "email": "jdoe@email.com"
        },
        "defaultAuthStrategy": "ALLOW",
        "groups": null,
        "issuer": "https://cognito-idp.us-west-2.amazonaws.com/us-west-xxxxxxxxxxx",
        "sourceIp": [
            "1.1.1.1"
        ],
        "sub": "192879fc-a240-4bf1-ab5a-d6a00f3063f9",
        "username": "jdoe"
    },
    "request": {
        "headers": {
            "x-amzn-trace-id": "Root=1-60488877-0b0c4e6727ab2a1c545babd0",
            "x-forwarded-for": "127.0.0.1",
            "cloudfront-viewer-country": "NL",
            "x-api-key": "da1-c33ullkbkze3jg5hf5ddgcs4fq"
        }
    }
}
Last update: 2022-10-05