AppSync onUpdate 订阅未由 python 请求触发 POST 到 GraphQL 端点

Question

我使用 Angular 创建了一个完整的堆栈应用程序，并使用 https://docs.amplify.aws/start/getting-started/data-model/q/integration/angular/ 中的教程进行了扩展。前端通过触发 onCreate 订阅的 AppSync 成功创建 GraphQL 条目，但从后端的 python lambda 函数更改该数据不会触发 onUpdate 订阅。

body.component.ts

  ngOnInit(): void {
        this.api.ListDestinationFiles().then((event) => {
          // do stuff here... (this works)
        });

        this.update_subscription = <Subscription>(
          this.api.OnUpdateDestinationFileListener(user.username).subscribe((event: any) => {
            // this never fires
            console.log('OnUpdate', event);
          }));
        this.create_subscription = <Subscription>(
          this.api.OnCreateDestinationFileListener(user.username).subscribe((event: any) => {
            // this does fire.
            console.log('OnCreate', event);
          }));
      })
  }

我的网页允许用户上传触发此功能的文件 body.component.ts:

  upload(): void {
    API.post(this.apiName, this.path, this.myInit)
      .then(response => {
        let keys = Object.keys(response['dest_file_urls']);
        this.graphql_api
          .CreateDestinationFile({
            filename: keys[0],
            signedURL: response['dest_file_urls'][keys[0]],
            status: 'Started'
          })
          .then((event) => {
            // this triggers successfully, and the item is indeed created
            console.log('item created', event);
          })

到目前为止一切都很好并且按预期工作。 API.post 将文件发送到 python lambda 函数，该函数将文件保存到 S3 存储桶，触发第二个 lambda 函数来完成处理文件的繁重工作，并且 returns 201给客户端（那是 'item created' 事件发生的时间）。

第二个 lambda 函数包含此代码：

destination_file = '{id: "' + str(gql_id) + '", filename: "' + d_filename + '", signedURL: "' + signedURL + '", status: "Finished"}'
mutation = 'mutation update {updateDestinationFile(input: ' + destination_file + ') { filename status }}'
response = session.request(url=endpoint, method='POST', headers=headers, json={'query': mutation})

AppSync POST 工作正常，数据正在后端更新，CloudWatch、Java 控制台等没有错误消息。唯一没有的是t 工作是触发 onUpdate 订阅，我有代码来更新网页，指示过程已完成。

我找到了各种关于为 AppSync 启用实时更新的文章和文档，其中包括添加或替换数据源类型为 NONE 的自动生成的解析器，以及进行各种传递，但这似乎是与最新版本的 AppSync 不再相关的遗留文档。我自动生成的更新解析器是一个 4 步管道处理身份验证、更新“updatedAt”字段等，并且在其中两个步骤中已经有了 NONE 数据源。

这是完整的架构，因为我有强烈的倾向，这是我需要以某种方式修改的内容，尽管它是自动生成的，我还没有找到明确的文档来在 schema.graphql 文件中进行修改.

input CreateDestinationFileInput {
    id: ID
    filename: String!
    signedURL: String
    status: String
}

input DeleteDestinationFileInput {
    id: ID!
}

type DestinationFile {
    id: ID!
    filename: String!
    signedURL: String
    status: String
    createdAt: AWSDateTime!
    updatedAt: AWSDateTime!
    owner: String
}

enum ModelAttributeTypes {
    binary
    binarySet
    bool
    list
    map
    number
    numberSet
    string
    stringSet
    _null
}

input ModelBooleanInput {
    ne: Boolean
    eq: Boolean
    attributeExists: Boolean
    attributeType: ModelAttributeTypes
}

input ModelDestinationFileConditionInput {
    filename: ModelStringInput
    signedURL: ModelStringInput
    status: ModelStringInput
    and: [ModelDestinationFileConditionInput]
    or: [ModelDestinationFileConditionInput]
    not: ModelDestinationFileConditionInput
}

type ModelDestinationFileConnection {
    items: [DestinationFile!]!
    nextToken: String
}

input ModelDestinationFileFilterInput {
    id: ModelIDInput
    filename: ModelStringInput
    signedURL: ModelStringInput
    status: ModelStringInput
    and: [ModelDestinationFileFilterInput]
    or: [ModelDestinationFileFilterInput]
    not: ModelDestinationFileFilterInput
}

input ModelFloatInput {
    ne: Float
    eq: Float
    le: Float
    lt: Float
    ge: Float
    gt: Float
    between: [Float]
    attributeExists: Boolean
    attributeType: ModelAttributeTypes
}

input ModelIDInput {
    ne: ID
    eq: ID
    le: ID
    lt: ID
    ge: ID
    gt: ID
    contains: ID
    notContains: ID
    between: [ID]
    beginsWith: ID
    attributeExists: Boolean
    attributeType: ModelAttributeTypes
    size: ModelSizeInput
}

input ModelIntInput {
    ne: Int
    eq: Int
    le: Int
    lt: Int
    ge: Int
    gt: Int
    between: [Int]
    attributeExists: Boolean
    attributeType: ModelAttributeTypes
}

input ModelSizeInput {
    ne: Int
    eq: Int
    le: Int
    lt: Int
    ge: Int
    gt: Int
    between: [Int]
}

enum ModelSortDirection {
    ASC
    DESC
}

input ModelStringInput {
    ne: String
    eq: String
    le: String
    lt: String
    ge: String
    gt: String
    contains: String
    notContains: String
    between: [String]
    beginsWith: String
    attributeExists: Boolean
    attributeType: ModelAttributeTypes
    size: ModelSizeInput
}

type Mutation {
    createDestinationFile(input: CreateDestinationFileInput!, condition: ModelDestinationFileConditionInput): DestinationFile
    updateDestinationFile(input: UpdateDestinationFileInput!, condition: ModelDestinationFileConditionInput): DestinationFile
    deleteDestinationFile(input: DeleteDestinationFileInput!, condition: ModelDestinationFileConditionInput): DestinationFile
}

type Query {
    getDestinationFile(id: ID!): DestinationFile
    listDestinationFiles(filter: ModelDestinationFileFilterInput, limit: Int, nextToken: String): ModelDestinationFileConnection
}

type Subscription {
    onCreateDestinationFile(owner: String): DestinationFile
        @aws_subscribe(mutations: ["createDestinationFile"])
    onUpdateDestinationFile(owner: String): DestinationFile
        @aws_subscribe(mutations: ["updateDestinationFile"])
    onDeleteDestinationFile(owner: String): DestinationFile
        @aws_subscribe(mutations: ["deleteDestinationFile"])
}

input UpdateDestinationFileInput {
    id: ID!
    filename: String
    signedURL: String
    status: String
}

这都是在 schema.graphql 文件中创建的：

type DestinationFile @model @auth(rules: [{ allow: owner }]) {
  id: ID!
  filename: String!
  signedURL: String
  status: String
}

Answer 1

您有没有启用 DataStore？如果你有那么它会为你所有的模型添加一个 _version 字段（类似于你提到的 updateAt 字段）。在 GraphQL 中进行任何更新时，您必须通过 _version 字段为该模型存储最后一个有效值，否则更新将失败（因此不会触发您的更新事件）。但这个理论只有在您实际启用了 DataStore 时才成立:)

Answer 2

我早该知道这会很傻。

mutation = 'mutation update {updateDestinationFile(input: ' + destination_file + ') { filename status }}'

return 集中缺少 owner 上面的行。由于我的模式包括：

type DestinationFile @model @auth(rules: [{ allow: owner }])

我必须 return owner 否则更新订阅永远看不到它。

mutation = 'mutation update {updateDestinationFile(input: ' + destination_file + ') { filename status owner }}

以上作品。

AppSync onUpdate 订阅未由 python 请求触发 POST 到 GraphQL 端点

AppSync onUpdate subscription not triggered by python requests POST to GraphQL endpoint

python

angular

aws-amplify

aws-appsync