AWS AppSync Construct Library
The aws-cdk-lib/aws-appsync
package contains constructs for building flexible
APIs that use GraphQL.
import aws_cdk.aws_appsync as appsync
Example
DynamoDB
Example of a GraphQL API with AWS_IAM
authorization resolving into a DynamoDb
backend data source.
GraphQL schema file schema.graphql
:
type demo {
id: String!
version: String!
}
type Query {
getDemos: [ demo! ]
}
input DemoInput {
version: String!
}
type Mutation {
addDemo(input: DemoInput!): demo
}
CDK stack file app-stack.ts
:
api = appsync.GraphqlApi(self, "Api",
name="demo",
definition=appsync.Definition.from_file(path.join(__dirname, "schema.graphql")),
authorization_config=appsync.AuthorizationConfig(
default_authorization=appsync.AuthorizationMode(
authorization_type=appsync.AuthorizationType.IAM
)
),
xray_enabled=True
)
demo_table = dynamodb.Table(self, "DemoTable",
partition_key=dynamodb.Attribute(
name="id",
type=dynamodb.AttributeType.STRING
)
)
demo_dS = api.add_dynamo_db_data_source("demoDataSource", demo_table)
# Resolver for the Query "getDemos" that scans the DynamoDb table and returns the entire list.
# Resolver Mapping Template Reference:
# https://docs.aws.amazon.com/appsync/latest/devguide/resolver-mapping-template-reference-dynamodb.html
demo_dS.create_resolver("QueryGetDemosResolver",
type_name="Query",
field_name="getDemos",
request_mapping_template=appsync.MappingTemplate.dynamo_db_scan_table(),
response_mapping_template=appsync.MappingTemplate.dynamo_db_result_list()
)
# Resolver for the Mutation "addDemo" that puts the item into the DynamoDb table.
demo_dS.create_resolver("MutationAddDemoResolver",
type_name="Mutation",
field_name="addDemo",
request_mapping_template=appsync.MappingTemplate.dynamo_db_put_item(
appsync.PrimaryKey.partition("id").auto(),
appsync.Values.projecting("input")),
response_mapping_template=appsync.MappingTemplate.dynamo_db_result_item()
)
# To enable DynamoDB read consistency with the `MappingTemplate`:
demo_dS.create_resolver("QueryGetDemosConsistentResolver",
type_name="Query",
field_name="getDemosConsistent",
request_mapping_template=appsync.MappingTemplate.dynamo_db_scan_table(True),
response_mapping_template=appsync.MappingTemplate.dynamo_db_result_list()
)
Aurora Serverless
AppSync provides a data source for executing SQL commands against Amazon Aurora Serverless clusters. You can use AppSync resolvers to execute SQL statements against the Data API with GraphQL queries, mutations, and subscriptions.
Aurora Serverless V1 Cluster
# Build a data source for AppSync to access the database.
# api: appsync.GraphqlApi
# Create username and password secret for DB Cluster
secret = rds.DatabaseSecret(self, "AuroraSecret",
username="clusteradmin"
)
# The VPC to place the cluster in
vpc = ec2.Vpc(self, "AuroraVpc")
# Create the serverless cluster, provide all values needed to customise the database.
cluster = rds.ServerlessCluster(self, "AuroraCluster",
engine=rds.DatabaseClusterEngine.AURORA_MYSQL,
vpc=vpc,
credentials={"username": "clusteradmin"},
cluster_identifier="db-endpoint-test",
default_database_name="demos"
)
rds_dS = api.add_rds_data_source("rds", cluster, secret, "demos")
# Set up a resolver for an RDS query.
rds_dS.create_resolver("QueryGetDemosRdsResolver",
type_name="Query",
field_name="getDemosRds",
request_mapping_template=appsync.MappingTemplate.from_string("""
{
"version": "2018-05-29",
"statements": [
"SELECT * FROM demos"
]
}
"""),
response_mapping_template=appsync.MappingTemplate.from_string("""
$utils.toJson($utils.rds.toJsonObject($ctx.result)[0])
""")
)
# Set up a resolver for an RDS mutation.
rds_dS.create_resolver("MutationAddDemoRdsResolver",
type_name="Mutation",
field_name="addDemoRds",
request_mapping_template=appsync.MappingTemplate.from_string("""
{
"version": "2018-05-29",
"statements": [
"INSERT INTO demos VALUES (:id, :version)",
"SELECT * WHERE id = :id"
],
"variableMap": {
":id": $util.toJson($util.autoId()),
":version": $util.toJson($ctx.args.version)
}
}
"""),
response_mapping_template=appsync.MappingTemplate.from_string("""
$utils.toJson($utils.rds.toJsonObject($ctx.result)[1][0])
""")
)
Aurora Serverless V2 Cluster
# Build a data source for AppSync to access the database.
# api: appsync.GraphqlApi
# Create username and password secret for DB Cluster
secret = rds.DatabaseSecret(self, "AuroraSecret",
username="clusteradmin"
)
# The VPC to place the cluster in
vpc = ec2.Vpc(self, "AuroraVpc")
# Create the serverless cluster, provide all values needed to customise the database.
cluster = rds.DatabaseCluster(self, "AuroraClusterV2",
engine=rds.DatabaseClusterEngine.aurora_postgres(version=rds.AuroraPostgresEngineVersion.VER_15_5),
credentials={"username": "clusteradmin"},
cluster_identifier="db-endpoint-test",
writer=rds.ClusterInstance.serverless_v2("writer"),
serverless_v2_min_capacity=2,
serverless_v2_max_capacity=10,
vpc=vpc,
default_database_name="demos",
enable_data_api=True
)
rds_dS = api.add_rds_data_source_v2("rds", cluster, secret, "demos")
# Set up a resolver for an RDS query.
rds_dS.create_resolver("QueryGetDemosRdsResolver",
type_name="Query",
field_name="getDemosRds",
request_mapping_template=appsync.MappingTemplate.from_string("""
{
"version": "2018-05-29",
"statements": [
"SELECT * FROM demos"
]
}
"""),
response_mapping_template=appsync.MappingTemplate.from_string("""
$utils.toJson($utils.rds.toJsonObject($ctx.result)[0])
""")
)
# Set up a resolver for an RDS mutation.
rds_dS.create_resolver("MutationAddDemoRdsResolver",
type_name="Mutation",
field_name="addDemoRds",
request_mapping_template=appsync.MappingTemplate.from_string("""
{
"version": "2018-05-29",
"statements": [
"INSERT INTO demos VALUES (:id, :version)",
"SELECT * WHERE id = :id"
],
"variableMap": {
":id": $util.toJson($util.autoId()),
":version": $util.toJson($ctx.args.version)
}
}
"""),
response_mapping_template=appsync.MappingTemplate.from_string("""
$utils.toJson($utils.rds.toJsonObject($ctx.result)[1][0])
""")
)
HTTP Endpoints
GraphQL schema file schema.graphql
:
type job {
id: String!
version: String!
}
input DemoInput {
version: String!
}
type Mutation {
callStepFunction(input: DemoInput!): job
}
GraphQL request mapping template request.vtl
:
{
"version": "2018-05-29",
"method": "POST",
"resourcePath": "/",
"params": {
"headers": {
"content-type": "application/x-amz-json-1.0",
"x-amz-target":"AWSStepFunctions.StartExecution"
},
"body": {
"stateMachineArn": "<your step functions arn>",
"input": "{ "id": "$context.arguments.id" }"
}
}
}
GraphQL response mapping template response.vtl
:
{
"id": "${context.result.id}"
}
CDK stack file app-stack.ts
:
api = appsync.GraphqlApi(self, "api",
name="api",
definition=appsync.Definition.from_file(path.join(__dirname, "schema.graphql"))
)
http_ds = api.add_http_data_source("ds", "https://states.amazonaws.com",
name="httpDsWithStepF",
description="from appsync to StepFunctions Workflow",
authorization_config=appsync.AwsIamConfig(
signing_region="us-east-1",
signing_service_name="states"
)
)
http_ds.create_resolver("MutationCallStepFunctionResolver",
type_name="Mutation",
field_name="callStepFunction",
request_mapping_template=appsync.MappingTemplate.from_file("request.vtl"),
response_mapping_template=appsync.MappingTemplate.from_file("response.vtl")
)
EventBridge
Integrating AppSync with EventBridge enables developers to use EventBridge rules to route commands for GraphQL mutations that need to perform any one of a variety of asynchronous tasks. More broadly, it enables teams to expose an event bus as a part of a GraphQL schema.
GraphQL schema file schema.graphql
:
schema {
query: Query
mutation: Mutation
}
type Query {
event(id:ID!): Event
}
type Mutation {
emitEvent(id: ID!, name: String): PutEventsResult!
}
type Event {
id: ID!
name: String!
}
type Entry {
ErrorCode: String
ErrorMessage: String
EventId: String
}
type PutEventsResult {
Entries: [Entry!]
FailedEntry: Int
}
GraphQL request mapping template request.vtl
:
{
"version" : "2018-05-29",
"operation": "PutEvents",
"events" : [
{
"source": "integ.appsync.eventbridge",
"detailType": "Mutation.emitEvent",
"detail": $util.toJson($context.arguments)
}
]
}
GraphQL response mapping template response.vtl
:
$util.toJson($ctx.result)'
This response mapping template simply converts the EventBridge PutEvents result to JSON. For details about the response see the documentation. Additional logic can be added to the response template to map the response type, or to error in the event of failed events. More information can be found here.
CDK stack file app-stack.ts
:
import aws_cdk.aws_events as events
api = appsync.GraphqlApi(self, "EventBridgeApi",
name="EventBridgeApi",
definition=appsync.Definition.from_file(path.join(__dirname, "appsync.eventbridge.graphql"))
)
bus = events.EventBus(self, "DestinationEventBus")
data_source = api.add_event_bridge_data_source("NoneDS", bus)
data_source.create_resolver("EventResolver",
type_name="Mutation",
field_name="emitEvent",
request_mapping_template=appsync.MappingTemplate.from_file("request.vtl"),
response_mapping_template=appsync.MappingTemplate.from_file("response.vtl")
)
Amazon OpenSearch Service
AppSync has builtin support for Amazon OpenSearch Service (successor to Amazon Elasticsearch Service) from domains that are provisioned through your AWS account. You can use AppSync resolvers to perform GraphQL operations such as queries, mutations, and subscriptions.
import aws_cdk.aws_opensearchservice as opensearch
# api: appsync.GraphqlApi
user = iam.User(self, "User")
domain = opensearch.Domain(self, "Domain",
version=opensearch.EngineVersion.OPENSEARCH_2_3,
removal_policy=RemovalPolicy.DESTROY,
fine_grained_access_control=opensearch.AdvancedSecurityOptions(master_user_arn=user.user_arn),
encryption_at_rest=opensearch.EncryptionAtRestOptions(enabled=True),
node_to_node_encryption=True,
enforce_https=True
)
ds = api.add_open_search_data_source("ds", domain)
ds.create_resolver("QueryGetTestsResolver",
type_name="Query",
field_name="getTests",
request_mapping_template=appsync.MappingTemplate.from_string(JSON.stringify({
"version": "2017-02-28",
"operation": "GET",
"path": "/id/post/_search",
"params": {
"headers": {},
"query_string": {},
"body": {"from": 0, "size": 50}
}
})),
response_mapping_template=appsync.MappingTemplate.from_string("""[
#foreach($entry in $context.result.hits.hits)
#if( $velocityCount > 1 ) , #end
$utils.toJson($entry.get("_source"))
#end
]""")
)
Merged APIs
AppSync supports Merged APIs which can be used to merge multiple source APIs into a single API.
import aws_cdk as cdk
# first source API
first_api = appsync.GraphqlApi(self, "FirstSourceAPI",
name="FirstSourceAPI",
definition=appsync.Definition.from_file(path.join(__dirname, "appsync.merged-api-1.graphql"))
)
# second source API
second_api = appsync.GraphqlApi(self, "SecondSourceAPI",
name="SecondSourceAPI",
definition=appsync.Definition.from_file(path.join(__dirname, "appsync.merged-api-2.graphql"))
)
# Merged API
merged_api = appsync.GraphqlApi(self, "MergedAPI",
name="MergedAPI",
definition=appsync.Definition.from_source_apis(
source_apis=[appsync.SourceApi(
source_api=first_api,
merge_type=appsync.MergeType.MANUAL_MERGE
), appsync.SourceApi(
source_api=second_api,
merge_type=appsync.MergeType.AUTO_MERGE
)
]
)
)
Merged APIs Across Different Stacks
The SourceApiAssociation construct allows you to define a SourceApiAssociation to a Merged API in a different stack or account. This allows a source API owner the ability to associate it to an existing Merged API itself.
source_api = appsync.GraphqlApi(self, "FirstSourceAPI",
name="FirstSourceAPI",
definition=appsync.Definition.from_file(path.join(__dirname, "appsync.merged-api-1.graphql"))
)
imported_merged_api = appsync.GraphqlApi.from_graphql_api_attributes(self, "ImportedMergedApi",
graphql_api_id="MyApiId",
graphql_api_arn="MyApiArn"
)
imported_execution_role = iam.Role.from_role_arn(self, "ExecutionRole", "arn:aws:iam::ACCOUNT:role/MyExistingRole")
appsync.SourceApiAssociation(self, "SourceApiAssociation2",
source_api=source_api,
merged_api=imported_merged_api,
merge_type=appsync.MergeType.MANUAL_MERGE,
merged_api_execution_role=imported_execution_role
)
Merge Source API Update Within CDK Deployment
The SourceApiAssociationMergeOperation construct available in the awscdk-appsync-utils package provides the ability to merge a source API to a Merged API via a custom resource. If the merge operation fails with a conflict, the stack update will fail and rollback the changes to the source API in the stack in order to prevent merge conflicts and ensure the source API changes are always propagated to the Merged API.
Custom Domain Names
For many use cases you may want to associate a custom domain name with your GraphQL API. This can be done during the API creation.
import aws_cdk.aws_certificatemanager as acm
import aws_cdk.aws_route53 as route53
# hosted zone and route53 features
# hosted_zone_id: str
zone_name = "example.com"
my_domain_name = "api.example.com"
certificate = acm.Certificate(self, "cert", domain_name=my_domain_name)
schema = appsync.SchemaFile(file_path="mySchemaFile")
api = appsync.GraphqlApi(self, "api",
name="myApi",
definition=appsync.Definition.from_schema(schema),
domain_name=appsync.DomainOptions(
certificate=certificate,
domain_name=my_domain_name
)
)
# hosted zone for adding appsync domain
zone = route53.HostedZone.from_hosted_zone_attributes(self, "HostedZone",
hosted_zone_id=hosted_zone_id,
zone_name=zone_name
)
# create a cname to the appsync domain. will map to something like xxxx.cloudfront.net
route53.CnameRecord(self, "CnameApiRecord",
record_name="api",
zone=zone,
domain_name=api.app_sync_domain_name
)
Log Group
AppSync automatically create a log group with the name /aws/appsync/apis/<graphql_api_id>
upon deployment with
log data set to never expire. If you want to set a different expiration period, use the logConfig.retention
property.
To obtain the GraphQL API’s log group as a logs.ILogGroup
use the logGroup
property of the
GraphqlApi
construct.
import aws_cdk.aws_logs as logs
log_config = appsync.LogConfig(
retention=logs.RetentionDays.ONE_WEEK
)
appsync.GraphqlApi(self, "api",
authorization_config=appsync.AuthorizationConfig(),
name="myApi",
definition=appsync.Definition.from_file(path.join(__dirname, "myApi.graphql")),
log_config=log_config
)
Schema
You can define a schema using from a local file using Definition.fromFile
api = appsync.GraphqlApi(self, "api",
name="myApi",
definition=appsync.Definition.from_file(path.join(__dirname, "schema.graphl"))
)
ISchema
Alternative schema sources can be defined by implementing the ISchema
interface. An example of this is the CodeFirstSchema
class provided in
awscdk-appsync-utils
Imports
Any GraphQL Api that has been created outside the stack can be imported from
another stack into your CDK app. Utilizing the fromXxx
function, you have
the ability to add data sources and resolvers through a IGraphqlApi
interface.
# api: appsync.GraphqlApi
# table: dynamodb.Table
imported_api = appsync.GraphqlApi.from_graphql_api_attributes(self, "IApi",
graphql_api_id=api.api_id,
graphql_api_arn=api.arn
)
imported_api.add_dynamo_db_data_source("TableDataSource", table)
If you don’t specify graphqlArn
in fromXxxAttributes
, CDK will autogenerate
the expected arn
for the imported api, given the apiId
. For creating data
sources and resolvers, an apiId
is sufficient.
Private APIs
By default all AppSync GraphQL APIs are public and can be accessed from the internet.
For customers that want to limit access to be from their VPC, the optional API visibility
property can be set to Visibility.PRIVATE
at creation time. To explicitly create a public API, the visibility
property should be set to Visibility.GLOBAL
.
If visibility is not set, the service will default to GLOBAL
.
CDK stack file app-stack.ts
:
api = appsync.GraphqlApi(self, "api",
name="MyPrivateAPI",
definition=appsync.Definition.from_file(path.join(__dirname, "appsync.schema.graphql")),
visibility=appsync.Visibility.PRIVATE
)
See documentation for more details about Private APIs
Permissions
When using AWS_IAM
as the authorization type for GraphQL API, an IAM Role
with correct permissions must be used for access to API.
When configuring permissions, you can specify specific resources to only be
accessible by IAM
authorization. For example, if you want to only allow mutability
for IAM
authorized access you would configure the following.
In schema.graphql
:
type Mutation {
updateExample(...): ...
@aws_iam
}
In IAM
:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"appsync:GraphQL"
],
"Resource": [
"arn:aws:appsync:REGION:ACCOUNT_ID:apis/GRAPHQL_ID/types/Mutation/fields/updateExample"
]
}
]
}
See documentation for more details.
To make this easier, CDK provides grant
API.
Use the grant
function for more granular authorization.
# api: appsync.IGraphqlApi
role = iam.Role(self, "Role",
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
api.grant(role, appsync.IamResource.custom("types/Mutation/fields/updateExample"), "appsync:GraphQL")
IamResource
In order to use the grant
functions, you need to use the class IamResource
.
IamResource.custom(...arns)
permits custom ARNs and requires an argument.IamResouce.ofType(type, ...fields)
permits ARNs for types and their fields.IamResource.all()
permits ALL resources.
Generic Permissions
Alternatively, you can use more generic grant
functions to accomplish the same usage.
These include:
grantMutation (use to grant access to Mutation fields)
grantQuery (use to grant access to Query fields)
grantSubscription (use to grant access to Subscription fields)
# api: appsync.IGraphqlApi
# role: iam.Role
# For generic types
api.grant_mutation(role, "updateExample")
# For custom types and granular design
api.grant(role, appsync.IamResource.of_type("Mutation", "updateExample"), "appsync:GraphQL")
Pipeline Resolvers and AppSync Functions
AppSync Functions are local functions that perform certain operations onto a backend data source. Developers can compose operations (Functions) and execute them in sequence with Pipeline Resolvers.
# api: appsync.GraphqlApi
appsync_function = appsync.AppsyncFunction(self, "function",
name="appsync_function",
api=api,
data_source=api.add_none_data_source("none"),
request_mapping_template=appsync.MappingTemplate.from_file("request.vtl"),
response_mapping_template=appsync.MappingTemplate.from_file("response.vtl")
)
AppSync Functions are used in tandem with pipeline resolvers to compose multiple operations.
# api: appsync.GraphqlApi
# appsync_function: appsync.AppsyncFunction
pipeline_resolver = appsync.Resolver(self, "pipeline",
api=api,
data_source=api.add_none_data_source("none"),
type_name="typeName",
field_name="fieldName",
request_mapping_template=appsync.MappingTemplate.from_file("beforeRequest.vtl"),
pipeline_config=[appsync_function],
response_mapping_template=appsync.MappingTemplate.from_file("afterResponse.vtl")
)
JS Functions and Resolvers
JS Functions and resolvers are also supported. You can use a .js
file within your CDK project, or specify your function code inline.
# api: appsync.GraphqlApi
my_js_function = appsync.AppsyncFunction(self, "function",
name="my_js_function",
api=api,
data_source=api.add_none_data_source("none"),
code=appsync.Code.from_asset("directory/function_code.js"),
runtime=appsync.FunctionRuntime.JS_1_0_0
)
appsync.Resolver(self, "PipelineResolver",
api=api,
type_name="typeName",
field_name="fieldName",
code=appsync.Code.from_inline("""
// The before step
export function request(...args) {
console.log(args);
return {}
}
// The after step
export function response(ctx) {
return ctx.prev.result
}
"""),
runtime=appsync.FunctionRuntime.JS_1_0_0,
pipeline_config=[my_js_function]
)
Learn more about Pipeline Resolvers and AppSync Functions here.
Introspection
By default, AppSync allows you to use introspection queries.
For customers that want to limit access to be introspection queries, the introspectionConfig
property can be set to IntrospectionConfig.DISABLED
at creation time.
If introspectionConfig
is not set, the service will default to ENABLED
.
api = appsync.GraphqlApi(self, "api",
name="DisableIntrospectionApi",
definition=appsync.Definition.from_file(path.join(__dirname, "appsync.schema.graphql")),
introspection_config=appsync.IntrospectionConfig.DISABLED
)
Query Depth Limits
By default, queries are able to process an unlimited amount of nested levels. Limiting queries to a specified amount of nested levels has potential implications for the performance and flexibility of your project.
api = appsync.GraphqlApi(self, "api",
name="LimitQueryDepths",
definition=appsync.Definition.from_file(path.join(__dirname, "appsync.schema.graphql")),
query_depth_limit=2
)
Resolver Count Limits
You can control how many resolvers each query can process. By default, each query can process up to 10000 resolvers. By setting a limit AppSync will not handle any resolvers past a certain number limit.
api = appsync.GraphqlApi(self, "api",
name="LimitResolverCount",
definition=appsync.Definition.from_file(path.join(__dirname, "appsync.schema.graphql")),
resolver_count_limit=2
)
Environment Variables
To use environment variables in resolvers, you can use the environmentVariables
property and
the addEnvironmentVariable
method.
api = appsync.GraphqlApi(self, "api",
name="api",
definition=appsync.Definition.from_file(path.join(__dirname, "appsync.schema.graphql")),
environment_variables={
"EnvKey1": "non-empty-1"
}
)
api.add_environment_variable("EnvKey2", "non-empty-2")