Uso de operações em lote do DynamoDB no AWS AppSync

O AppSync da AWS oferece suporte ao uso de operações em lote do Amazon DynamoDB em uma ou mais tabelas em uma única região. As operações compatíveis são BatchGetItem, BatchPutItem e BatchDeleteItem. Ao usar esses atributos no AWS AppSync, execute tarefas como:

As operações em lote no AWS AppSync apresentam duas diferenças importantes em relação às operações que não são em lote:

Lotes de tabela única

Para começar, vamos criar uma API GraphQL. No console AWS AppSync, escolha Criar API, APIs GraphQL e Design do zero. Nomeie sua API BatchTutorial API, escolha Próximo e, na etapa Especificar recursos do GraphQL, selecione Criar recursos do GraphQL mais tarde e clique em Próximo. Revise seus detalhes e crie a API. Vá para a página Esquema e cole o esquema a seguir, observando que, para a consulta, enviaremos uma lista de IDs:

type Post {
    id: ID!
    title: String
}

input PostInput {
    id: ID!
    title: String
}

type Query {
    batchGet(ids: [ID]): [Post]
}

type Mutation {
    batchAdd(posts: [PostInput]): [Post]
    batchDelete(ids: [ID]): [Post]
}

Salve seu esquema e escolha Criar recursos na parte superior da página. Escolha Usar tipo existente e selecione o tipo Post. Nomeie sua tabela Posts. Certifique-se de que a Chave primária esteja definida como id, desmarque Gerar GraphQL automaticamente (você fornecerá seu próprio código) e selecione Criar. Para começar, AWS AppSync cria uma tabela do DynamoDB e uma fonte de dados conectada à tabela com as funções apropriadas. No entanto, ainda há algumas permissões que você precisa adicionar à função. Acesse a página Fontes de dados e escolha a nova fonte de dados. Em Selecionar uma função existente, você notará que uma função foi criada automaticamente para a tabela. Anote a função (deve ser semelhante a appsync-ds-ddb-aaabbbcccddd-Posts) e vá para o console do IAM (https://console.aws.amazon.com/iam/). No console do IAM, escolha Funções e selecione sua função na tabela. Na sua função, em Políticas de permissões, clique em "+" ao lado da política (deve ter um nome semelhante ao nome da função). Escolha Editar na parte superior do expansível quando a política aparece. Você precisa adicionar permissões em lote à sua política, especificamente dynamodb:BatchGetItem e dynamodb:BatchWriteItem. Essa lista será semelhante a:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "dynamodb:DeleteItem",
                "dynamodb:GetItem",
                "dynamodb:PutItem",
                "dynamodb:Query",
                "dynamodb:Scan",
                "dynamodb:UpdateItem",
                "dynamodb:BatchWriteItem",
                "dynamodb:BatchGetItem"
            ],
            "Resource": [
                "arn:aws:dynamodb:…",
                "arn:aws:dynamodb:…"
            ]
        }
    ]
}

Escolha Avançar e Salvar alterações. Sua política deve permitir o processamento em lote agora.

De volta ao console AWS AppSync, acesse a página Esquema e selecione Anexar ao lado do campo Mutation.batchAdd. Crie seu resolvedor usando a tabela Posts como fonte de dados. No editor de código, substitua os manipuladores pelo trecho abaixo. Esse trecho recebe automaticamente cada item no tipo input PostInput do GraphQL e cria um mapa, que é necessário para a operação BatchPutItem:

import { util } from "@aws-appsync/utils";

export function request(ctx) {
  return {
    operation: "BatchPutItem",
    tables: {
      Posts: ctx.args.posts.map((post) => util.dynamodb.toMapValues(post)),
    },
  };
}

export function response(ctx) {
  if (ctx.error) {
    util.error(ctx.error.message, ctx.error.type);
  }
  return ctx.result.data.Posts;
}

Navegue até a página Consultas do console do AWSAppSync e execute a mutação batchAdd a seguir:

mutation add {
    batchAdd(posts:[{
            id: 1 title: "Running in the Park"},{
            id: 2 title: "Playing fetch"
        }]){
            id
            title
    }
}

Você deverá ver os resultados impressos na tela; isso pode ser validado revisando o console do DynamoDB para verificar os valores gravados na tabela Posts.

Em seguida, repita o processo de anexar um resolvedor, mas para o campo Query.batchGet usando a tabela Posts como fonte de dados. Substitua os manipuladores pelo código abaixo. Isso recebe automaticamente cada item no tipo ids:[] do GraphQL e cria um mapa, que é necessário para a operação BatchGetItem:

import { util } from "@aws-appsync/utils";

export function request(ctx) {
  return {
    operation: "BatchGetItem",
    tables: {
      Posts: {
        keys: ctx.args.ids.map((id) => util.dynamodb.toMapValues({ id })),
        consistentRead: true,
      },
    },
  };
}

export function response(ctx) {
  if (ctx.error) {
    util.error(ctx.error.message, ctx.error.type);
  }
  return ctx.result.data.Posts;
}

Agora volte à página Consultas do console do AWS AppSync e execute a seguinte consulta batchGet:

query get {
    batchGet(ids:[1,2,3]){
        id
        title
    }
}

Isso deve retornar os resultados para os dois valores id adicionados anteriormente. Observe que um valor null foi retornado para o id com um valor de 3. Isso ocorre porque não havia registro na tabela Posts com esse valor ainda. Observe também que o AWS AppSync retorna os resultados na mesma ordem que as chaves enviadas à consulta, que é um atributo adicional realizado pelo AWS AppSync para você. Portanto, se você alternar para batchGet(ids:[1,3,2]), verá a ordem alterada. Você também saberá qual id retornou um valor null.

Por fim, anexe mais um resolvedor ao campo Mutation.batchDelete usando a tabela Posts como fonte de dados.​ Substitua os manipuladores pelo código abaixo. Isso recebe automaticamente cada item no tipo ids:[] do GraphQL e cria um mapa, que é necessário para a operação BatchGetItem:

import { util } from "@aws-appsync/utils";

export function request(ctx) {
  return {
    operation: "BatchDeleteItem",
    tables: {
      Posts: ctx.args.ids.map((id) => util.dynamodb.toMapValues({ id })),
    },
  };
}

export function response(ctx) {
  if (ctx.error) {
    util.error(ctx.error.message, ctx.error.type);
  }
  return ctx.result.data.Posts;
}

Agora volte à página Consultas do console do AWS AppSync e execute a seguinte mutação batchDelete:

mutation delete {
    batchDelete(ids:[1,2]){ id }
}

Os registros com id, 1 e 2 devem ser excluídos agora. Se você executar novamente a consulta batchGet() de anteriormente, eles devem retornar null.

Lote de várias tabelas

O AWS AppSync também permite realizar operações em lote para tabelas. Vamos criar um aplicativo mais complexo. Imagine que estamos criando um aplicativo de saúde para animais de estimação em que sensores informam a localização e a temperatura corporal do animal. Os sensores são alimentados por bateria e tentam se conectar à rede a cada cinco minutos. Quando um sensor estabelece conexão, ele envia as leituras para a nossa API do AWS AppSync. Em seguida, os gatilhos analisam os dados para que um painel seja apresentado ao dono do animal. Vamos nos concentrar na representação das interações entre o sensor e o armazenamento de dados de back-end.

No console AWS AppSync, escolha Criar API, APIs GraphQL e Design do zero. Nomeie sua API MultiBatchTutorial API, escolha Próximo e, na etapa Especificar recursos do GraphQL, selecione Criar recursos do GraphQL mais tarde e clique em Próximo. Revise seus detalhes e crie a API. Vá até a página Esquema, cole e salve o seguinte esquema:

type Mutation {
    # Register a batch of readings
    recordReadings(tempReadings: [TemperatureReadingInput], locReadings: [LocationReadingInput]): RecordResult
    # Delete a batch of readings
    deleteReadings(tempReadings: [TemperatureReadingInput], locReadings: [LocationReadingInput]): RecordResult
}

type Query {
    # Retrieve all possible readings recorded by a sensor at a specific time
    getReadings(sensorId: ID!, timestamp: String!): [SensorReading]
}

type RecordResult {
    temperatureReadings: [TemperatureReading]
    locationReadings: [LocationReading]
}

interface SensorReading {
    sensorId: ID!
    timestamp: String!
}

# Sensor reading representing the sensor temperature (in Fahrenheit)
type TemperatureReading implements SensorReading {
    sensorId: ID!
    timestamp: String!
    value: Float
}

# Sensor reading representing the sensor location (lat,long)
type LocationReading implements SensorReading {
    sensorId: ID!
    timestamp: String!
    lat: Float
    long: Float
}

input TemperatureReadingInput {
    sensorId: ID!
    timestamp: String
    value: Float
}

input LocationReadingInput {
    sensorId: ID!
    timestamp: String
    lat: Float
    long: Float
}

Precisamos criar duas tabelas do DynamoDB:

Ambas as tabelas compartilharão a mesma estrutura de chave primária: sensorId (String) como chave de partição e timestamp (String) como chave de classificação.

Escolha Criar recursos na parte superior da página. Escolha Usar tipo existente e selecione o tipo locationReadings. Nomeie sua tabela locationReadings. Certifique-se de que a Chave primária esteja definida como sensorId e a chave de classificação como timestamp. Desmarque Gerar GraphQL automaticamente (você fornecerá seu próprio código) e selecione Criar. Repita esse processo para temperatureReadings usar temperatureReadings como tipo e nome da tabela. Use as mesmas teclas acima.

Suas novas tabelas conterão funções geradas automaticamente. Ainda existem algumas permissões que você precisa adicionar a essas funções. Vá para a página Fontes de dados e escolha locationReadings. Em Selecionar uma função existente, é possível ver a função. Anote a função (deve ser semelhante a appsync-ds-ddb-aaabbbcccddd-locationReadings) e vá para o console do IAM (https://console.aws.amazon.com/iam/). No console do IAM, escolha Funções e selecione sua função na tabela. Na sua função, em Políticas de permissões, clique em "+" ao lado da política (deve ter um nome semelhante ao nome da função). Escolha Editar na parte superior do expansível quando a política aparece. Você precisa adicionar permissões a essa política. Essa lista será semelhante a:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "dynamodb:DeleteItem",
                "dynamodb:GetItem",
                "dynamodb:PutItem",
                "dynamodb:Query",
                "dynamodb:Scan",
                "dynamodb:UpdateItem",
                "dynamodb:BatchGetItem",
                "dynamodb:BatchWriteItem"
            ],
            "Resource": [
                "arn:aws:dynamodb:region:account:table/locationReadings",
                "arn:aws:dynamodb:region:account:table/locationReadings/*",
                "arn:aws:dynamodb:region:account:table/temperatureReadings",
                "arn:aws:dynamodb:region:account:table/temperatureReadings/*"
            ]
        }
    ]
}

Escolha Avançar e Salvar alterações. Repita esse processo para a fonte de dados temperatureReadings usando o mesmo trecho de política acima.

BatchPutItem – gravação de leituras do sensor

Nossos sensores precisam ser capazes de enviar suas leituras assim que se conectarem à internet. O campo do GraphQL Mutation.recordReadings é a API que será usada para fazer isso. Precisaremos adicionar um resolvedor a esse campo.

Na página Esquema do console AWS AppSync, selecione Anexar ao lado do campo Mutation.recordReadings. Na próxima tela, crie seu resolvedor usando a tabela locationReadings como fonte de dados.

Depois de criar seu resolvedor, substitua os manipuladores pelo código a seguir no editor. Essa operação BatchPutItem nos permite especificar múltiplas tabelas:

import { util } from '@aws-appsync/utils'

export function request(ctx) {
	const { locReadings, tempReadings } = ctx.args
	const locationReadings = locReadings.map((loc) => util.dynamodb.toMapValues(loc))
	const temperatureReadings = tempReadings.map((tmp) => util.dynamodb.toMapValues(tmp))

	return {
		operation: 'BatchPutItem',
		tables: {
			locationReadings,
			temperatureReadings,
		},
	}
}

export function response(ctx) {
	if (ctx.error) {
		util.appendError(ctx.error.message, ctx.error.type)
	}
	return ctx.result.data
}

Com operações em lote, podem haver erros e resultados retornados na invocação. Nesse caso, podemos fazer uma manipulação de erros adicional.

Salve o resolvedor e navegue até a página Consultas do console do AWS AppSync. Agora podemos enviar algumas leituras do sensor.

Execute a seguinte mutação:

mutation sendReadings {
  recordReadings(
    tempReadings: [
      {sensorId: 1, value: 85.5, timestamp: "2018-02-01T17:21:05.000+08:00"},
      {sensorId: 1, value: 85.7, timestamp: "2018-02-01T17:21:06.000+08:00"},
      {sensorId: 1, value: 85.8, timestamp: "2018-02-01T17:21:07.000+08:00"},
      {sensorId: 1, value: 84.2, timestamp: "2018-02-01T17:21:08.000+08:00"},
      {sensorId: 1, value: 81.5, timestamp: "2018-02-01T17:21:09.000+08:00"}
    ]
    locReadings: [
      {sensorId: 1, lat: 47.615063, long: -122.333551, timestamp: "2018-02-01T17:21:05.000+08:00"},
      {sensorId: 1, lat: 47.615163, long: -122.333552, timestamp: "2018-02-01T17:21:06.000+08:00"},
      {sensorId: 1, lat: 47.615263, long: -122.333553, timestamp: "2018-02-01T17:21:07.000+08:00"},
      {sensorId: 1, lat: 47.615363, long: -122.333554, timestamp: "2018-02-01T17:21:08.000+08:00"},
      {sensorId: 1, lat: 47.615463, long: -122.333555, timestamp: "2018-02-01T17:21:09.000+08:00"}
    ]) {
    locationReadings {
      sensorId
      timestamp
      lat
      long
    }
    temperatureReadings {
      sensorId
      timestamp
      value
    }
  }
}

Enviamos dez leituras do sensor em uma mutação, com as leituras divididas em duas tabelas. Use o console do DynamoDB para validar se os dados aparecem nas tabelas locationReadings e temperatureReadings.

BatchDeleteItem – exclusão de leituras do sensor

Da mesma forma, também será necessário excluir lotes de leituras do sensor. Vamos usar o campo do GraphQL Mutation.deleteReadings para essa finalidade. Na página Esquema do console AWS AppSync, selecione Anexar ao lado do campo Mutation.deleteReadings. Na próxima tela, crie seu resolvedor usando a tabela locationReadings como fonte de dados.

Depois de criar seu resolvedor, substitua os manipuladores no editor de código pelo trecho abaixo. Neste resolvedor, usamos um mapeador de função auxiliar que extrai sensorId e timestamp das entradas fornecidas.

import { util } from '@aws-appsync/utils'

export function request(ctx) {
	const { locReadings, tempReadings } = ctx.args
	const mapper = ({ sensorId, timestamp }) => util.dynamodb.toMapValues({ sensorId, timestamp })

	return {
		operation: 'BatchDeleteItem',
		tables: {
			locationReadings: locReadings.map(mapper),
			temperatureReadings: tempReadings.map(mapper),
		},
	}
}

export function response(ctx) {
	if (ctx.error) {
		util.appendError(ctx.error.message, ctx.error.type)
	}
	return ctx.result.data
}

Salve o resolvedor e navegue até a página Consultas do console do AWS AppSync. Agora, vamos excluir algumas leituras do sensor.

Execute a seguinte mutação:

mutation deleteReadings {
  # Let's delete the first two readings we recorded
  deleteReadings(
    tempReadings: [{sensorId: 1, timestamp: "2018-02-01T17:21:05.000+08:00"}]
    locReadings: [{sensorId: 1, timestamp: "2018-02-01T17:21:05.000+08:00"}]) {
    locationReadings {
      sensorId
      timestamp
      lat
      long
    }
    temperatureReadings {
      sensorId
      timestamp
      value
    }
  }
}

Valide por meio do console do DynamoDB que essas duas leituras foram excluídas das tabelas locationReadings e temperatureReadings.

BatchGetItem – recuperar leituras

Outra operação comum para o nosso aplicativo seria recuperar as leituras de um sensor em um determinado momento. Vamos anexar um resolvedor ao campo do GraphQL Query.getReadings em nosso esquema. Na página Esquema do console AWS AppSync, selecione Anexar ao lado do campo Query.getReadings. Na próxima tela, crie seu resolvedor usando a tabela locationReadings como fonte de dados.

Vamos usar o seguinte código:

import { util } from '@aws-appsync/utils'

export function request(ctx) {
	const keys = [util.dynamodb.toMapValues(ctx.args)]
	const consistentRead = true
	return {
		operation: 'BatchGetItem',
		tables: {
			locationReadings: { keys, consistentRead },
			temperatureReadings: { keys, consistentRead },
		},
	}
}

export function response(ctx) {
	if (ctx.error) {
		util.appendError(ctx.error.message, ctx.error.type)
	}
	const { locationReadings: locs, temperatureReadings: temps } = ctx.result.data

	return [
		...locs.map((l) => ({ ...l, __typename: 'LocationReading' })),
		...temps.map((t) => ({ ...t, __typename: 'TemperatureReading' })),
	]
}

Salve o resolvedor e navegue até a página Consultas do console do AWS AppSync. Agora, vamos recuperar as leituras dos nossos sensores.

Execute a seguinte consulta:

query getReadingsForSensorAndTime {
  # Let's retrieve the very first two readings
  getReadings(sensorId: 1, timestamp: "2018-02-01T17:21:06.000+08:00") {
    sensorId
    timestamp
    ...on TemperatureReading {
      value
    }
    ...on LocationReading {
      lat
      long
    }
  }
}

Demonstramos com sucesso o uso de operações em lote do DynamoDB usando o AWS AppSync.

Tratamento de erros

No AWS AppSync, as operações da fonte de dados podem, às vezes, retornar resultados parciais. Resultados parciais será o termo usado para indicar quando a saída de uma operação for composta por alguns dados e um erro. Como a manipulação de erros é inerentemente específica ao aplicativo, o AWS AppSync oferece a oportunidade de lidar com erros no manipulador de resposta. O erro de invocação do resolvedor, se presente, está disponível no contexto como ctx.error. Os erros de invocação sempre incluem uma mensagem e um tipo, acessível como propriedades ctx.error.message e ctx.error.type. No manipulador de respostas, é possível manipular resultados parciais de três maneiras:

Vamos demonstrar cada um dos três pontos acima com operações em lote do DynamoDB!

Operações em lote do DynamoDB

Com as operações em lote do DynamoDB, é possível que um lote seja concluído parcialmente. Ou seja, é possível que alguns dos itens solicitados ou chaves não seja processados. Se o AWS AppSync não conseguir concluir um lote, os itens não processados e um erro de invocação serão definidos no contexto.

Vamos implementar a manipulação de erros usando a configuração de campo Query.getReadings da operação BatchGetItem da seção anterior desse tutorial. Dessa vez, vamos fingir que ao executar o campo Query.getReadings, a tabela do DynamoDB temperatureReadings esgotou sua throughput provisionada. O DynamoDB gerou um ProvisionedThroughputExceededException durante a segunda tentativa do AWS AppSync em processar os elementos restantes no lote.

O JSON a seguir representa o contexto serializado após a invocação em lote do DynamoDB, mas antes do manipulador de resposta ser chamado:

{
  "arguments": {
    "sensorId": "1",
    "timestamp": "2018-02-01T17:21:05.000+08:00"
  },
  "source": null,
  "result": {
    "data": {
      "temperatureReadings": [
        null
      ],
      "locationReadings": [
        {
          "lat": 47.615063,
          "long": -122.333551,
          "sensorId": "1",
          "timestamp": "2018-02-01T17:21:05.000+08:00"
        }
      ]
    },
    "unprocessedKeys": {
      "temperatureReadings": [
        {
          "sensorId": "1",
          "timestamp": "2018-02-01T17:21:05.000+08:00"
        }
      ],
      "locationReadings": []
    }
  },
  "error": {
    "type": "DynamoDB:ProvisionedThroughputExceededException",
    "message": "You exceeded your maximum allowed provisioned throughput for a table or for one or more global secondary indexes. (...)"
  },
  "outErrors": []
}

Algumas observações sobre o contexto:

Vamos lidar com esse erro de três maneiras diferentes.

1. Absorção do erro de invocação

Retornar dados sem manipular o erro de invocação efetivamente absorve o erro, tornando o resultado para o determinado campo do GraphQL sempre bem-sucedido.

O código que gravamos é familiar e concentra-se apenas nos dados do resultado.

Manipulador de resposta

export function response(ctx) {
  return ctx.result.data
}

Resposta do GraphQL

{
  "data": {
    "getReadings": [
      {
        "sensorId": "1",
        "timestamp": "2018-02-01T17:21:05.000+08:00",
        "lat": 47.615063,
        "long": -122.333551
      },
      {
        "sensorId": "1",
        "timestamp": "2018-02-01T17:21:05.000+08:00",
        "value": 85.5
      }
    ]
  }
}

Nenhum erro será adicionado à resposta do erro uma vez que apenas dados foram modificados.

2. Gerar um erro para abortar a execução do manipulador de resposta

Quando falhas parciais devem ser tratadas como falhas completas do ponto de vista do cliente, você pode abortar a execução do manipulador de respostas para evitar o retorno de dados. O método utilitário util.error(...) atinge exatamente esse comportamento.

Código do manipulador de respostas

export function response(ctx) {
  if (ctx.error) {
    util.error(ctx.error.message, ctx.error.type, null, ctx.result.data.unprocessedKeys);
  }
  return ctx.result.data;
}

Resposta do GraphQL

{
  "data": {
    "getReadings": null
  },
  "errors": [
    {
      "path": [
        "getReadings"
      ],
      "data": null,
      "errorType": "DynamoDB:ProvisionedThroughputExceededException",
      "errorInfo": {
        "temperatureReadings": [
          {
            "sensorId": "1",
            "timestamp": "2018-02-01T17:21:05.000+08:00"
          }
        ],
        "locationReadings": []
      },
      "locations": [
        {
          "line": 58,
          "column": 3
        }
      ],
      "message": "You exceeded your maximum allowed provisioned throughput for a table or for one or more global secondary indexes. (...)"
    }
  ]
}

Embora alguns resultados possam ter sido retornados da operação em lote do DynamoDB, escolhemos gerar um erro tal que o campo do GraphQL getReadings seja nulo e o erro seja adicionado ao bloco erros da resposta do GraphQL.

3. Anexar um erro para retornar dados e erros

Em alguns casos, para oferecer uma experiência melhor ao usuário, os aplicativos podem retornar resultados parciais e notificar seus clientes sobre os itens não processados. Os clientes podem decidir implementar uma nova tentativa ou traduzir o erro de volta para o usuário final. O util.appendError(...) é o método utilitário que permite esse comportamento, permitindo que o designer do aplicativo anexe erros no contexto sem interferir na avaliação do manipulador de respostas. Depois de avaliar o manipulador de respostas, o AWS AppSync processará qualquer erro de contexto anexando-os ao bloco de erros da resposta do GraphQL.

Código do manipulador de respostas

export function response(ctx) {
  if (ctx.error) {
    util.appendError(ctx.error.message, ctx.error.type, null, ctx.result.data.unprocessedKeys);
  }
  return ctx.result.data;
}

Encaminhamos o erro de invocação e o elemento unprocessedKeys dentro do bloco de erros da resposta do GraphQL. O campo getReadings também retorna dados parciais da tabela locationReadings como você pode ver na resposta abaixo.

Resposta do GraphQL

{
  "data": {
    "getReadings": [
      null,
      {
        "sensorId": "1",
        "timestamp": "2018-02-01T17:21:05.000+08:00",
        "value": 85.5
      }
    ]
  },
  "errors": [
    {
      "path": [
        "getReadings"
      ],
      "data": null,
      "errorType": "DynamoDB:ProvisionedThroughputExceededException",
      "errorInfo": {
        "temperatureReadings": [
          {
            "sensorId": "1",
            "timestamp": "2018-02-01T17:21:05.000+08:00"
          }
        ],
        "locationReadings": []
      },
      "locations": [
        {
          "line": 58,
          "column": 3
        }
      ],
      "message": "You exceeded your maximum allowed provisioned throughput for a table or for one or more global secondary indexes. (...)"
    }
  ]
}