Skip to content

Appsync

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.

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!
    title: String
    description: String
    done: 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
125
126
127
128
129
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Hello world Direct Lambda Resolver

Globals:
  Function:
    Timeout: 5
    Runtime: python3.8
    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: sample_resolver

Resources:
  HelloWorldFunction:
    Type: AWS::Serverless::Function
    Properties:
        Handler: app.lambda_handler
        CodeUri: hello_world
        Description: Sample Lambda Powertools Direct Lambda Resolver
        Tags:
            SOLUTION: LambdaPowertoolsPython

  # 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 HelloWorldFunction.Arn
      Roles:
          - !Ref AppSyncServiceRole

  # GraphQL API

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

  HelloWorldApiKey:
    Type: AWS::AppSync::ApiKey
    Properties:
        ApiId: !GetAtt HelloWorldApi.ApiId

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

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

            type Todo {
                id: ID!
                title: String
                description: String
                done: Boolean
            }

  # Lambda Direct Data Source and Resolver

  HelloWorldFunctionDataSource:
    Type: "AWS::AppSync::DataSource"
    Properties:
        ApiId: !GetAtt HelloWorldApi.ApiId
        Name: "HelloWorldLambdaDirectResolver"
        Type: "AWS_LAMBDA"
        ServiceRoleArn: !GetAtt AppSyncServiceRole.Arn
        LambdaConfig:
            LambdaFunctionArn: !GetAtt HelloWorldFunction.Arn

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

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


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

  HelloWorldAPI:
    Value: !GetAtt HelloWorldApi.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.

GraphQL arguments are passed as function arguments

 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
from aws_lambda_powertools import Logger, Tracer

from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.utilities.data_classes.appsync import scalar_types_utils

tracer = Tracer(service="sample_resolver")
logger = Logger(service="sample_resolver")
app = AppSyncResolver()

# Note that `creation_time` isn't available in the schema
# This utility also takes into account what info you make available at API level vs what's stored
TODOS = [
    {
        "id": scalar_types_utils.make_id(), # type ID or String
        "title": "First task",
        "description": "String",
        "done": False,
        "creation_time": scalar_types_utils.aws_datetime(),  # type AWSDateTime
    },
    {
        "id": scalar_types_utils.make_id(),
        "title": "Second task",
        "description": "String",
        "done": True,
        "creation_time": scalar_types_utils.aws_datetime(),
    },
]


@app.resolver(type_name="Query", field_name="getTodo")
def get_todo(id: str = ""):
    logger.info(f"Fetching Todo {id}")
    todo = [todo for todo in TODOS if todo["id"] == id]

    return todo


@app.resolver(type_name="Query", field_name="listTodos")
def list_todos():
    return TODOS


@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
@tracer.capture_lambda_handler
def lambda_handler(event, context):
    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!
    title: String
    description: String
    done: 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": {}
}

Advanced

Nested mappings

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
  from aws_lambda_powertools import Logger, Tracer

  from aws_lambda_powertools.logging import correlation_paths
  from aws_lambda_powertools.event_handler import AppSyncResolver

  tracer = Tracer(service="sample_resolver")
  logger = Logger(service="sample_resolver")
  app = AppSyncResolver()

  @app.resolver(field_name="listLocations")
  @app.resolver(field_name="locations")
  def get_locations(name: str, description: str = ""):
      return name + description

  @logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
  @tracer.capture_lambda_handler
  def lambda_handler(event, context):
      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: [Todo]
}

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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
from aws_lambda_powertools import Logger, Tracer

from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.event_handler import AppSyncResolver

tracer = Tracer(service="sample_resolver")
logger = Logger(service="sample_resolver")
app = AppSyncResolver()

@app.resolver(type_name="Query", field_name="listTodos")
async def list_todos():
    todos = await some_async_io_call()
    return todos

@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)
@tracer.capture_lambda_handler
def lambda_handler(event, context):
    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.

 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.

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 Lambda 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
from aws_lambda_powertools import Logger, Tracer

from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.utilities.data_classes.appsync import scalar_types_utils

tracer = Tracer(service="sample_graphql_transformer_resolver")
logger = Logger(service="sample_graphql_transformer_resolver")
app = AppSyncResolver()

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

@app.resolver(field_name="commonField")
def common_field():
    # 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, context):
    app.resolve(event, context)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
from aws_lambda_powertools.event_handler import AppSyncResolver
from aws_lambda_powertools.utilities.data_classes.appsync import scalar_types_utils

app = AppSyncResolver()

@app.resolver(type_name="Query", field_name="findMerchant")
def find_merchant(search: str):
    return [
      {
        "id": scalar_types_utils.make_id(),
        "name": "Brewer Brewing",
        "description": "Mike Brewer's IPA brewing place"
      },
      {
        "id": scalar_types_utils.make_id(),
        "name": "Serverlessa's Bakery",
        "description": "Lessa's sourdough place"
      },
    ]

Example AppSync GraphQL Transformer Function resolver events

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
{
  "typeName": "Query",
  "fieldName": "listLocations",
  "arguments": {
    "page": 2,
    "size": 1
  },
  "identity": {
    "claims": {
      "iat": 1615366261
      ...
    },
    "username": "mike",
    ...
  },
  "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
20
21
22
{
  "typeName": "Merchant",
  "fieldName": "commonField",
  "arguments": {
  },
  "identity": {
    "claims": {
      "iat": 1615366261
      ...
    },
    "username": "mike",
    ...
  },
  "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
20
21
22
23
{
  "typeName": "Query",
  "fieldName": "findMerchant",
  "arguments": {
    "search": "Brewers Coffee"
  },
  "identity": {
    "claims": {
      "iat": 1615366261
      ...
    },
    "username": "mike",
    ...
  },
  "request": {
    "headers": {
      "x-amzn-trace-id": "Root=1-60488877-0b0c4e6727ab2a1c545babd0",
      "x-forwarded-for": "127.0.0.1"
      ...
    }
  },
  ...
}

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 from our internal functional test.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
def test_direct_resolver():
  # Check whether we can handle an example appsync direct resolver
  # load_event primarily deserialize the JSON event into a dict
  mock_event = load_event("appSyncDirectResolver.json")

  app = AppSyncResolver()

  @app.resolver(field_name="createSomething")
  def create_something(id: str):
      assert app.lambda_context == {}
      return id

  # Call the implicit handler
  result = app(mock_event, {})

  assert result == "my identifier"

 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
{
  "arguments": {
    "id": "my identifier"
  },
  "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"
  },
  "source": null,
  "request": {
    "headers": {
      "x-forwarded-for": "1.1.1.1, 2.2.2.2",
      "cloudfront-viewer-country": "US",
      "cloudfront-is-tablet-viewer": "false",
      "via": "2.0 xxxxxxxxxxxxxxxx.cloudfront.net (CloudFront)",
      "cloudfront-forwarded-proto": "https",
      "origin": "https://us-west-1.console.aws.amazon.com",
      "content-length": "217",
      "accept-language": "en-US,en;q=0.9",
      "host": "xxxxxxxxxxxxxxxx.appsync-api.us-west-1.amazonaws.com",
      "x-forwarded-proto": "https",
      "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.83 Safari/537.36",
      "accept": "*/*",
      "cloudfront-is-mobile-viewer": "false",
      "cloudfront-is-smarttv-viewer": "false",
      "accept-encoding": "gzip, deflate, br",
      "referer": "https://us-west-1.console.aws.amazon.com/appsync/home?region=us-west-1",
      "content-type": "application/json",
      "sec-fetch-mode": "cors",
      "x-amz-cf-id": "3aykhqlUwQeANU-HGY7E_guV5EkNeMMtwyOgiA==",
      "x-amzn-trace-id": "Root=1-5f512f51-fac632066c5e848ae714",
      "authorization": "eyJraWQiOiJScWFCSlJqYVJlM0hrSnBTUFpIcVRXazNOW...",
      "sec-fetch-dest": "empty",
      "x-amz-user-agent": "AWS-Console-AppSync/",
      "cloudfront-is-desktop-viewer": "true",
      "sec-fetch-site": "cross-site",
      "x-forwarded-port": "443"
    }
  },
  "prev": null,
  "info": {
    "selectionSetList": [
      "id",
      "field1",
      "field2"
    ],
    "selectionSetGraphQL": "{\n  id\n  field1\n  field2\n}",
    "parentTypeName": "Mutation",
    "fieldName": "createSomething",
    "variables": {}
  },
  "stash": {}
}
Last update: 2021-04-09