使用AWS CDK Python构建AWS AppSync API

2 年 ago
清, 宇
4 minutes

首先

我在以下文章中研究了使用 AWS CDK Python 进行部署的方法。本次记录了使用 AWS CDK Python 实际部署 AWS AppSync 的方法。

有关AWS CDK Python的设置方法，请参考上述文章。

执行环境等

    • Ubuntu 20.04 LTS

 

    • aws-cdk: 1.125.0 (build 67b4921)

 

    • pipenv: version 2021.5.29

 

    デプロイ先: ap-northeast-1 (Tokyo) リージョン

部署 AWS AppSync

本文介绍如何部署以下简单的GraphQL API架构。

type User {
  name: String!
  age: Int!
}

type Query {
  hello: String
  getUsers: [ User! ]
}

type Mutation {
  addUser(name: String!, age: Int!): User
}

准备好

为了将DynamoDB用作GraphQL API的数据源，并将AWS Lambda用作返回固定字符串的数据源，因此需要在Python中安装各个服务的库以使用它们。

$ pipenv install aws-cdk.aws-appsync aws-cdk.aws-dynamodb aws-cdk.aws-lambda

AWS AppSync 服务的结构

在写CDK代码之前，先简要解释一下AppSync的结构。因为在编写代码时了解结构，可以更容易理解。主要内容包括Schema、DataSource和Resolvers三部分，它们都位于GraphQL API之下。

GraphQL API: トップレベルコンテンツであり、エンドポイントを持つ

Schema: GraphQL API に紐づくスキーマ定義

DataSource: AWS AppSyncが連携する (AppSync外の) サービスの紐づけの定義。 例えば、どのような DynamoDB Table を利用するか、など

Resolvers: Schema と DataSource を結びつけるルール定義。 Schema の入力を DataSource で定義されている各サービスで利用できる形に変換してやる必要があるため、この変換・対応を実現する部分

How-It-Works-product-page-diagram_AppSync@2x.c39ecb89b7b40ba682cf0f875ae13b6bdd8dca5b.png

这幅图是从AWS官方引用来的，但是从这个图中可以看出，DataSource可以独立于AppSync存在。AppSync只是作为获取和操作这些数据源的外部服务存在。

AppSync Stack 的实现

在使用cdk init app进行初始化后，我编辑和添加了以下内容。带有(*)的内容是我新添加的。
在使用cdk init app进行初始化后，我编辑和添加了以下内容。带有(*)的是我新增加的。
使用cdk init app进行初始化后，我编辑和添加了以下内容。有(*)标记的是我新增加的。

.
├── cdk_appsync
│   └── cdk_appsync_stack.py
├── lambda  (*)
│   └── fixed_string.py  (*)
├── resolvers  (*)
│   └── addUser.vtl  (*)
├── schema.graphql  (*)

模式.graphql

这里是对所创建的 GraphQL API 的模式定义进行记录。
由于这与之前的模式文件相同，所以可以省略。

lambda/fixed_string.py => lambda/fixed_string.py

这是一个用作AppSync数据源的Lambda函数，返回固定的字符串。

def handler(event, context):
    return 'This is a fixed string!'

添加用户的解析器功能模板文件：resolvers/addUser.vtl

AppSync的解析器可以使用Apache Velocity模板语言（VTL）进行编写。后文将介绍，虽然可以使用CDK中提供的助手来编写解析器，但这里为了展示使用VTL的示例而做准备。

内容如下：

{
    "version" : "2017-02-28",
    "operation" : "PutItem",
    "key" : {
        "name": $util.dynamodb.toDynamoDBJson($ctx.args.name)
    },
    "attributeValues" : {
        "age": $util.dynamodb.toDynamoDBJson($ctx.args.age)
    }
}

这是一个实施 DynamoDB 映射模板并实现 putItem 功能的例子。还可以适用于其他 DynamoDB 的不同操作（如 GetItem 或 ListItem）。请参考本文来了解具体的写法。

cdk_appsync/cdk_appsync_stack.py 可以被改写成中文为：cdk_appsync/cdk_appsync_stack.py

这里是正文。请参考以下页面等进行记录。

#!/usr/bin/python
# -*- coding: utf-8 -*-

from aws_cdk import (
    core as cdk,
    aws_lambda as cdk_lambda,
    aws_appsync as cdk_appsync,
    aws_dynamodb as cdk_dynamodb
)
from aws_cdk.aws_appsync import (
    AuthorizationConfig, AuthorizationMode, AuthorizationType,
    MappingTemplate
)


class CdkAppSyncStack(cdk.Stack):

    def __init__(self, scope: cdk.Construct,
                 construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # GraphQL API と利用するスキーマの定義
        graph_api = cdk_appsync.GraphqlApi(
            self, 'HelloAppSyncFromCDK',
            name='HelloAppSyncFromCDK',
            authorization_config=AuthorizationConfig(
                default_authorization=AuthorizationMode(
                    authorization_type=AuthorizationType.API_KEY
                )
            ),
            schema=cdk_appsync.Schema.from_asset('schema.graphql'),
            xray_enabled=False
        )

        # DynamoDB DataSource を追加
        data_table = cdk_dynamodb.Table(
            self, "DemoTable",
            partition_key=cdk_dynamodb.Attribute(
                name="name",
                type=cdk_dynamodb.AttributeType.STRING
            )
        )
        data_source = graph_api.add_dynamo_db_data_source(
            'demoDataSource', data_table)

        # getUsers に全件取得の Resolver を付与 (プログラムベース)
        data_source.create_resolver(
            type_name='Query',
            field_name='getUsers',
            request_mapping_template=MappingTemplate.dynamo_db_scan_table(),
            response_mapping_template=MappingTemplate.dynamo_db_result_list()
        )

        # addUser に vtl テンプレートの Resolver を付与 (VTLベース)
        data_source.create_resolver(
            type_name='Mutation',
            field_name='addUser',
            request_mapping_template=MappingTemplate.from_file(
                'resolvers/addUser.vtl'
            ),
            response_mapping_template=MappingTemplate.dynamo_db_result_item()
        )

        # hello に固定文字列を返す Lambda 関数の Resolver を付与
        fixed_str = cdk_lambda.Function(
            self, 'FixedStringLambda',
            runtime=cdk_lambda.Runtime.PYTHON_3_8,
            code=cdk_lambda.Code.asset('lambda'),
            handler='fixed_string.handler',
            environment={},
        )
        graph_api.add_lambda_data_source(
            'FixedStringFn', fixed_str
        ).create_resolver(
            type_name='Query', field_name='hello'
        )

通过代码，我们可以清楚地看出以下特点：”数据源的外部资源定义（如 data_table 和 fixed_str）”，”与外部资源的关联（add_xxxxxxx_data_source）”，”与解析器的对应关系定义（create_resolver）”都被明确地分开了。我认为，如果我们理解了AppSync的结构和定位，这些就会迅速地融入我们的思维。

如果实施到这一步，可以使用pipenv run cdk deploy进行部署。

部署后的操作验证

尝试使用curl进行操作验证。由于这次进行了基于API密钥的安全措施，所以我们需要从AppScyn控制台获取终端点URL和API密钥来使用。

# 固定文字列を取得するクエリ: hello を発行
$ curl -s -X POST -H "Context-Type: application/json" /
       -H "X-API-Key: ************************************" \
       -d '{"query": "{ hello }"}' \
       https://**************************.appsync-api.ap-northeast-1.amazonaws.com/graphql | jq .
{
  "data": {
    "hello": "This is a fixed string!"
  }
}

# ユーザーを追加する mutation 
$ curl -s -X POST -H "Context-Type: application/json" /
       -H "X-API-Key: ************************************" \
       -d '{"query": "mutation { addUser(name: \"TestUser1\", age: 20) { name age } }"}' \
       https://**************************.appsync-api.ap-northeast-1.amazonaws.com/graphql | jq .
{
  "data": {
    "addUser": {
      "name": "TestUser1",
      "age": 20
    }
  }
}

# 全ユーザー取得するクエリ
$ curl -s -X POST -H "Context-Type: application/json" /
       -H "X-API-Key: ************************************" \
       -d '{"query": "{ getUsers { name age } }"}' \
       https://**************************.appsync-api.ap-northeast-1.amazonaws.com/graphql | jq . 
{
  "data": {
    "getUsers": [
      {
        "name": "TestUser1",
        "age": 20
      }
    ]
  }
}

# --------
# 再度ユーザーを追加
$ curl -s -X POST -H "Context-Type: application/json" /
       -H "X-API-Key: ************************************" \
       -d '{"query": "mutation { addUser(name: \"TestUser2\", age: 30) { name age } }"}' \
       https://**************************.appsync-api.ap-northeast-1.amazonaws.com/graphql | jq .
{
  "data": {
    "addUser": {
      "name": "TestUser2",
      "age": 30
    }
  }
}

# 全ユーザー取得するクエリ
$ curl -s -X POST -H "Context-Type: application/json" /
       -H "X-API-Key: ************************************" \
       -d '{"query": "{ getUsers { name age } }"}' \
       https://**************************.appsync-api.ap-northeast-1.amazonaws.com/graphql | jq . 
{
  "data": {
    "getUsers": [
      {
        "name": "TestUser1",
        "age": 20
      },
      {
        "name": "TestUser2",
        "age": 30
      }
    ]
  }
}

另外，还能确认数据已保存在DynamoDB表中。

SnapCrab_NoName_2021-10-3_17-33-44_No-00.png

总结

我在這裡使用AWS CDK Python來定義並部署了一個簡單的AppSync API。同時，我也確認了使用端點進行的操作。基本上，如果基於AWS Management Console，它提供了一些常用的模板自動生成功能，甚至可以從模式定義中自動生成數據源和解析器等功能。這樣，我可以在控制台上輕鬆手動創建資源，同時將必要的部分切割並轉移到AWS CDK中，使工作更順利進行。

bannerAds