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!
    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.

Info

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: [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
import asyncio
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.

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 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"
      ...
    }
  },
  ...
}

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
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_resolver_event import AppSyncResolverEvent

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


class MyCustomModel(AppSyncResolverEvent):
    @property
    def country_viewer(self) -> str:
        return self.request_headers.get("cloudfront-viewer-country")

@app.resolver(field_name="listLocations")
@app.resolver(field_name="locations")
def get_locations(name: str, description: str = ""):
    if app.current_event.country_viewer == "US":
      ...
    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, 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
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
{
  "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",
      "name",
      "description"
    ],
    "selectionSetGraphQL": "{\n  id\n  name\n  description\n}",
    "fieldName": "listLocations",
    "variables": {}
  },
  "stash": {}
}

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 where the Router feature is useful.

Let's assume you have app.py as your Lambda function entrypoint and routes in location.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
from typing import Any, Dict, List

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

logger = Logger(child=True)
router = Router()


@router.resolver(type_name="Query", field_name="listLocations")
def list_locations(merchant_id: str) -> List[Dict[str, Any]]:
    return [{"name": "Location name", "merchant_id": merchant_id}]


@router.resolver(type_name="Location", field_name="status")
def resolve_status(merchant_id: str) -> str:
    logger.debug(f"Resolve status for merchant_id: {merchant_id}")
    return "FOO"

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
18
19
from typing import Dict

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

from resolvers import location

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


@tracer.capture_lambda_handler
@logger.inject_lambda_context(correlation_id_path=APPSYNC_RESOLVER)
def lambda_handler(event: Dict, context: LambdaContext):
    app.resolve(event, context)

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
import json
import pytest
from pathlib import Path

from src.index import app  # import the instance of AppSyncResolver from your code

def test_direct_resolver():
  # Load mock event from a file
  json_file_path = Path("appSyncDirectResolver.json")
  with open(json_file_path) as json_file:
    mock_event = json.load(json_file)

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

  assert result == "created this value"

1
2
3
4
5
6
7
from aws_lambda_powertools.event_handler import AppSyncResolver

app = AppSyncResolver()

@app.resolver(field_name="createSomething")
def create_something():
    return "created this 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
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": {}
}

And an example for testing asynchronous resolvers. Note that this requires the pytest-asyncio package:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
import json
import pytest
from pathlib import Path

from src.index import app  # import the instance of AppSyncResolver from your code

@pytest.mark.asyncio
async def test_direct_resolver():
  # Load mock event from a file
  json_file_path = Path("appSyncDirectResolver.json")
  with open(json_file_path) as json_file:
    mock_event = json.load(json_file)

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

  assert result == "created this value"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import asyncio

from aws_lambda_powertools.event_handler import AppSyncResolver

app = AppSyncResolver()

@app.resolver(field_name="createSomething")
async def create_something_async():
    await asyncio.sleep(1)  # Do async stuff
    return "created this 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
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: 2022-03-17
Back to top