AWS AppSync에서 GraphQL 해석기 결합

포커스 모드

AWS AppSync에서 GraphQL 해석기 결합 - AWS AppSync GraphQL

스키마 예제 해석기를 통해 데이터 변경 DynamoDB 및 OpenSearch Service

GraphQL 스키마의 해석기와 필드는 매우 뛰어난 유연성으로 1:1 관계를 갖습니다. 데이터 소스는 스키마와 독립적으로 해석기에 구성되므로, 다양한 데이터 소스를 통해 GraphQL 유형을 확인하거나 조작할 수 있으므로 필요에 따라 스키마를 혼합하고 일치시킬 수 있습니다.

다음 시나리오는 스키마에서 데이터 소스를 혼합하고 일치시키는 방법을 보여줍니다. 시작하기 전에, Amazon DynamoDB 및 AWS Lambda Amazon OpenSearch Service에 대한 데이터 소스 및 해석기 구성에 익숙해야 합니다.

스키마 예제

다음 스키마에는 각각 3개의 Query 및 Mutation 작업이 있는 Post 유형이 있습니다.


type Post {
    id: ID!
    author: String!
    title: String
    content: String
    url: String
    ups: Int
    downs: Int
    version: Int!
}

type Query {
    allPost: [Post]
    getPost(id: ID!): Post
    searchPosts: [Post]
}

type Mutation {
    addPost(
        id: ID!,
        author: String!,
        title: String,
        content: String,
        url: String
    ): Post
    updatePost(
        id: ID!,
        author: String!,
        title: String,
        content: String,
        url: String,
        ups: Int!,
        downs: Int!,
        expectedVersion: Int!
    ): Post
    deletePost(id: ID!): Post
}

이 예제에서는 각각 데이터 소스가 필요한 총 6개의 해석기가 있습니다. 이 문제를 해결하는 한 가지 방법은 Posts라는 단일 Amazon DynamoDB 테이블에 연결하여 AllPost 필드는 스캔을 실행하고 searchPosts 필드는 쿼리를 실행하는 것입니다(DynamoDB용 JavaScript 해석기 함수 참조 참조). 그러나 Amazon DynamoDB에만 국한되지는 않습니다. 비즈니스 요구 사항을 충족하기 위해 Lambda 또는 OpenSearch Service와 같은 다양한 데이터 소스가 존재합니다.

해석기를 통해 데이터 변경

AWS AppSync 데이터 소스에서 직접 지원하지 않는 타사 데이터베이스의 결과를 반환해야 할 수 있습니다. 또한 데이터를 API 클라이언트에 반환하기 전에 복잡한 수정을 수행해야 할 수도 있습니다. 이는 클라이언트의 타임스탬프 차이 또는 이전 버전과의 호환성 문제 처리와 같은 데이터 형식의 부적절한 서식으로 인해 발생할 수 있습니다. 이 경우 AWS Lambda 함수를 AWS AppSync API에 데이터 소스로 연결하는 것이 적절한 솔루션입니다. 예시를 위해 다음 예제에서 AWS Lambda 함수는 타사 데이터 스토어에서 가져온 데이터를 조작합니다.


export const handler = (event, context, callback) => {
    // fetch data
    const result = fetcher()

    // apply complex business logic
    const data = transform(result)	

    // return to AppSync
    return data
};

이 함수는 완벽하게 유효한 Lambda 함수로 GraphQL 스키마의 AllPost 필드에 연결하여 모든 결과에서 반환하는 쿼리가 좋아요/싫어요에 대한 난수를 가져오도록 합니다.

DynamoDB 및 OpenSearch Service

일부 애플리케이션의 경우 DynamoDB에 대한 단순 조회 쿼리와 변형을 수행할 수 있으며, 백그라운드 프로세스를 통해 문서를 OpenSearch Service로 전송할 수 있습니다. 단순히 searchPosts 해석기를 OpenSearch Service 데이터 소스에 연결하고 GraphQL 쿼리를 사용하여 (DynamoDB에서 가져온 데이터의) 검색 결과를 반환할 수 있습니다. 이 기능은 키워드, 퍼지 워드 일치 또는 지역 검색 조회 등 애플리케이션에 고급 검색 작업 추가 시 매우 유용할 수 있습니다. DynamoDB에서 데이터 전송은 ETL 프로세스를 통해 수행하거나 Lambda을 사용하여 DynamoDB에서 스트리밍할 수 있습니다.

이러한 특정 데이터 소스로 시작하려면 DynamoDB 및 Lambda 자습서를 참조하세요.

예를 들어, 이전 튜토리얼의 스키마를 사용하여 다음 뮤테이션을 수행하면 DynamoDB에 항목이 추가됩니다.


mutation addPost {
  addPost(
    id: 123
    author: "Nadia"
    title: "Our first post!"
    content: "This is our first post."
    url: "https://aws.amazon.com/appsync/"
  ) {
    id
    author
    title
    content
    url
    ups
    downs
    version
  }
}

이렇게 하면 DynamoDB에 데이터가 기록되고, Lambda를 통해 Amazon OpenSearch Service로 데이터를 스트리밍합니다. 그러면 이를 사용하여 다양한 필드로 게시물을 검색할 수 있습니다. 예를 들면 데이터가 Amazon OpenSearch Service에 있으므로 다음과 같이 공백으로만 이루어졌더라도 자유 형식의 텍스트를 사용하여 작성자나 콘텐츠 필드를 검색할 수 있습니다.


query searchName{
    searchAuthor(name:"   Nadia   "){
        id
        title
        content
    }
}

---------- or ----------

query searchContent{
    searchContent(text:"test"){
        id
        title
        content
    }
}

데이터가 DynamoDB에 직접 기록되므로 allPost{...} 및 getPost{...} 쿼리를 사용하여 테이블에 대해 효율적인 목록이나 항목 조회 작업을 수행할 수 있습니다. 이 스택은 DynamoDB 스트림에 대해 다음 예제 코드를 사용합니다.

참고

이 Python 코드는 예시이며 프로덕션 코드에 사용하기 위한 것이 아닙니다.


import boto3
import requests
from requests_aws4auth import AWS4Auth

region = '' # e.g. us-east-1
service = 'es'
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)

host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com
index = 'lambda-index'
datatype = '_doc'
url = host + '/' + index + '/' + datatype + '/'

headers = { "Content-Type": "application/json" }

def handler(event, context):
    count = 0
    for record in event['Records']:
        # Get the primary key for use as the OpenSearch ID
        id = record['dynamodb']['Keys']['id']['S']

        if record['eventName'] == 'REMOVE':
            r = requests.delete(url + id, auth=awsauth)
        else:
            document = record['dynamodb']['NewImage']
            r = requests.put(url + id, auth=awsauth, json=document, headers=headers)
        count += 1
    return str(count) + ' records processed.'

DynamoDB 스트림을 사용하여 id를 기본 키로 하는 DynamoDB 테이블에 이 부분을 연결하면, 원본에 대한 모든 변경 내용이 OpenSearch Service 도메인으로 스트리밍됩니다. 이 구성에 대한 자세한 내용은 DynamoDB Streams 설명서를 참조하십시오.