Nodejs续集批量更新

Question

有没有办法在 sequelize 中进行批量 upsert。另外，我可以指定用于检查重复项的键吗？

我尝试了以下方法，但没有成功：

Employee.bulkCreate(data, {
    updateOnDuplicate: true
});

虽然批量创建工作正常。上面的语句总是在数据库中创建新条目。

Answer 1

来自官方sequelizejs reference。

可以使用bulkCreate 和updateOnDuplicate 选项来完成。

例如这样：

Employee.bulkCreate(dataArray, 
    {
        fields:["id", "name", "address"] ,
        updateOnDuplicate: ["name"] 
    } )

updateOnDuplicate 是一个字段数组，当主键（或可能是唯一键）匹配行时将更新这些字段。确保您的模型和 dataArray 中至少有一个唯一字段（比如说 id）用于 upsert。

Answer 2

更新

Sequelize 6.x 增加了对所有方言的所有 UPSERT 的支持，所以@followtest52 的回答对 PostgreSQL 也有效。

原创

由于答案不支持 PostgreSQL，使用 Sequelize 的 """"best"""" 替代方案是使用 ON CONFLICT 语句进行手动查询。示例（打字稿）：

const values: Array<Array<number | string>> = [
    [1, 'Apple', 'Red', 'Yummy'],
    [2, 'Kiwi', 'Green', 'Yuck'],
]

const query = 'INSERT INTO fruits (id, name, color, flavor) VALUES ' +
     values.map(_ => { return '(?)' }).join(',') +
     ' ON CONFLICT (id) DO UPDATE SET flavor = excluded.flavor;'

sequelize.query({ query, values }, { type: sequelize.QueryTypes.INSERT })

这将构建如下查询：

INSERT INTO 
    fruits (id, name, color, flavor)
VALUES 
    (1, 'Apple', 'Red', 'Yummy'),
    (2, 'Kiwi', 'Green', 'Yuck')
ON CONFLICT (id) DO UPDATE SET 
    flavor = excluded.flavor;

可以说，这不是必须手动构建查询的理想解决方案，因为它违背了使用 sequelize 的目的，但如果它是一次性查询，您并不迫切需要，您可以使用此方法。

Answer 3

2019 年更新

适用于所有方言，只要匹配某个最低版本

HERE是对相同源代码的引用

• 请注意，个别选项可能适用于所有方言，也可能不适用于所有方言 例如，updateOnDuplicate 仅适用于 MySQL、MariaDB、 SQLite 和 Postgres

• ignoreDuplicates 选项不适用于 MSSQL

还要检查源代码中的这块代码

if (Array.isArray(options.updateOnDuplicate) && options.updateOnDuplicate.length) {
    options.updateOnDuplicate = _.intersection(
        _.without(Object.keys(model.tableAttributes), createdAtAttr),
        options.updateOnDuplicate
    );
} else {
    return Promise.reject(new Error('updateOnDuplicate option only supports non-empty array.'));
}

updateOnDuplicate 必须是一个数组，不能为真或假

按照以上几点，你的代码应该是这样的

Employee.bulkCreate(data, {
    updateOnDuplicate: ['employeeName', 'employeeAge'],
});

更新：

既然有人提到它不起作用，试试这个

models.Employee.bulkCreate(items, {
    returning: ['employeeId'],
    ignoreDuplicates: true
  })

Answer 4

2020 年 10 月 1 日更新
续集版本：^6.3.5

问题仍然存在。我们仍然无法使用唯一的复合索引 bulkUpsert。 bulkCreate 和 updateOnDuplicates 还不能使用唯一的复合索引。仍有 PR 等待合并，可能会解决此问题：-
https://github.com/sequelize/sequelize/pull/12516
https://github.com/sequelize/sequelize/pull/12547

解决方法

目前，如果有人想要快速解决方法，那么可以通过修改您自己的表属性、名称和数据来使用以下基于原始查询的包装器：-

const bulkUpsertIntoTable = async ({ bulkUpsertableData }) => {
  try {
    /* eslint-disable */
   // id column will automatically be incremented if you have set it to auto-increment
   const query = `INSERT INTO "Table" ("non_id_attr1", "non_id_attr2", "non_id_attr3","createdAt", "updatedAt") VALUES ${bulkUpsertableData
    .map((_) => "(?)")
    .join(
      ","
    )} ON CONFLICT ("non_id_attr1","non_id_attr2") DO UPDATE SET "non_id_attr1"=excluded."non_id_attr1", "non_id_attr2"=excluded."non_id_attr2", "non_id_attr3"=excluded."non_id_attr3",  "updatedAt"=excluded."updatedAt" RETURNING "id","non_id_attr1","non_id_attr2","non_id_attr3","createdAt","updatedAt";`;
    /* eslint-enable */

    return await models.sequelize.query(query, {
      replacements: bulkUpsertableData,//------> dont forget to pass your data here
      type: models.Sequelize.QueryTypes.INSERT,
      // transaction:t -----> if required to be done in transaction
    });
  } catch (error) {
    console.error("Bulk Upserting into Table:", error);
    throw error;
  }
};

重点是创建bulkUpsertableData，它应该是Array<Array> ie:- [[]]。示例创建：-

// with reference to above wrapper function
const bulkUpsertableData = Object.keys(myObjectData).map(type => [
      myObjectData[type],// -----> non_id_attr1
      type, // -----> non_id_attr2
      someOtherRandomValue, // -----> non_id_attr3
      new Date(), // -----> created_at
      new Date(), // -----> updated_at
]);

// response will have all the raw attributes mentioned in RETURNING clause
const upsertedTableResponse = await bulkUpsertIntoTable({ bulkUpsertableData });

Answer 5

2020 年 11 月 2 日更新

根据@Yedhin 的回答，这里有一个更通用的解决方案（打字稿）：

export const bulkUpsert = async <T extends Model<T>, K extends keyof T>(
  items: Partial<T>[],
  model: ModelCtor<T>,
  conflictKeys: K[],
  excludeFromUpdate: K[] = [],
): Promise<[number, number]> => {
  if (!items.length) {
    return [0, 0];
  }

  const { tableName, sequelize, name } = model;
  if (!sequelize) {
    throw new Error(`Sequelize not initialized on ${name}?`);
  }

  const sample = items[0];
  const fields = Object.keys(sample) as K[];
  const createFields = `("${fields.join(`","`)}")`;
  const updateFields = fields
    .filter((field) => ![...excludeFromUpdate, ...conflictKeys].includes(field))
    .map((field) => `"${field}"=EXCLUDED."${field}"`)
    .join(', ');
  const values = items.map(dataToSql(sequelize)).join(',');
  const onConflict = `ON CONFLICT ("${conflictKeys.join(`","`)}")`;
  const returning = `"${fields.join('","')}"`;

  const query = `INSERT INTO "${tableName}" ${createFields} VALUES ${values} ${onConflict} DO UPDATE SET ${updateFields} RETURNING ${returning};`;

  return sequelize.query(query, {
    replacements: items,
    type: QueryTypes.INSERT,
  });
};

const valueToSql = (sequelize: Sequelize) => (
  value: string | number | boolean | null | Date | string[] | Record<string, unknown>,
): string => {
  if (value === null) {
    return 'null';
  }

  if (typeof value === 'boolean') {
    return value ? 'true' : 'false';
  }

  if (typeof value !== 'object' || value instanceof Date) {
    return sequelize.escape(value);
  }

  return sequelize.escape(JSON.stringify(value));
};


const dataToSql = <T extends Node<T>>(sequelize: Sequelize) => (data: Partial<T>): string =>
  `(${Object.values(data).map(valueToSql(sequelize)).join(',')})`;

Answer 6

2021 年 9 月更新

具有唯一复合索引的批量更新插入现在仅适用于 Sequelize v6.4.4。

https://github.com/sequelize/sequelize/pull/13345

Answer 7

修改后的版本。哪个可以完成这项工作。

/**
 *
 * @param {*} data Raw JSON data
 * @param {*} model Sequalize model
 * @param {*} fields Columns thare need to be inserted/update.If none passed, it will extract fields from the data.
 * @returns response consists of data with type of action(upsert/create) performed for each record.
 */
export const bulkUpert = (data, model, fields = undefined) => {
  console.log("****Bulk insertion started****");
  if (!data.length) {
    return [0, 0];
  }
  const { name, primaryKeyAttributes } = model;

  console.log(name, primaryKeyAttributes, fields);

  if (!sequelize) {
    throw new Error(`Sequalize not initialized on ${name}`);
  }

  const extractFields = fields ? fields : Object.keys(data[0]);
  const createFields = extractFields.join(", ");
  const values = data.map(dataToSql()).join(", ");

  const query = `MERGE INTO
    [${name}]
    WITH(HOLDLOCK)
    AS [targetTable]
    USING (
        VALUES ${values}
    )
    AS [sourceTable]
    (
      ${createFields}
    ) ON
    ${getPrimaryQueryString(primaryKeyAttributes)}
    WHEN MATCHED THEN
        UPDATE SET
            ${getUpdateFieldsString(extractFields)}
    WHEN NOT MATCHED THEN
        INSERT (
              ${createFields}
            )
        VALUES
            (
                ${getInsertValuesString(extractFields)}
            )
    OUTPUT $action, INSERTED.*;`;
  return sequelize.query(query);
};

const valueToSQL = () => (value) => {
  if (value === null) {
    return "null";
  }

  if (typeof value === "boolean") {
    return value ? "true" : "false";
  }

  if (typeof value !== "object" || value instanceof Date) {
    return sequelize.escape(value);
  }

  return sequelize.escape(JSON.stringify(value));
};

const getPrimaryQueryString = (primaryKeyAttributes) => {
  let string = "";
  for (let i = 0; i < primaryKeyAttributes.length; i++) {
    string += `[targetTable].[${primaryKeyAttributes[i]}] = [sourceTable].[${primaryKeyAttributes[i]}]`;
    if (i != primaryKeyAttributes.length - 1) {
      string += " AND";
    }
  }
  return string;
};

const getUpdateFieldsString = (fields) => {
  let string = "";
  for (let i = 0; i < fields.length; i++) {
    string += `[targetTable].[${fields[i]}] = [sourceTable].[${fields[i]}]`;
    if (i != fields.length - 1) {
      string += ", ";
    }
  }
  return string;
};

const getInsertValuesString = (fields) => {
  let string = "";
  for (let i = 0; i < fields.length; i++) {
    string += `[sourceTable].[${fields[i]}]`;
    if (i != fields.length - 1) {
      string += ", ";
    }
  }
  return string;
};

const dataToSql = () => (data) =>
  `(${Object.values(data).map(valueToSQL()).join(",")})`;