批量更新插入存储过程

Question

如果有人可以提供帮助，我是 cosmos 和 JS 的新手。我有一个需要批量更新数据的要求，我为此找到了 JavaScript，但这失败了。

尝试从 python 调用存储过程：

client.ExecuteStoredProcedure(sproc_linkOut, [new_docs,True], options = options) 

或

client.ExecuteStoredProcedure(sproc_linkOut, [new_docs], options = options)

但我在下面显示错误。但我认为它不应该因为错误 409 而出现，这段代码是写的。你能帮忙吗？

HTTPFailure：状态码：400 子状态：409
{"code":"BadRequest",
"message":"Message: {"Errors":["执行函数时遇到异常。异常 = 错误：{\"Errors\":[\"具有指定 ID 或名称的资源已存在。\"]}\r\n堆栈跟踪：错误：{\"Errors\":[\"具有指定 ID 或名称的资源名称已存在。\"]}。

JavaScript 存储过程：

//------------------------------------------------------------
// Copyright (c) Microsoft Corporation. All rights reserved.
//------------------------------------------------------------

function bulkImport(docs, upsert) {
    var collection = getContext().getCollection();
    var collectionLink = collection.getSelfLink();

    // The count of imported docs, also used as current doc index.
    var count = 0;
    var errorCodes = { CONFLICT: 409 };

    // Validate input.
    if (!docs) throw new Error("The array is undefined or null.");

    var docsLength = docs.length;

    if (docsLength == 0) {
        getContext().getResponse().setBody(0);
        return;
    }

    // Call the create API to create a document.
    tryCreate(docs[count], callback);

    // Note that there are 2 exit conditions:
    // 1) The createDocument request was not accepted.
    // In this case the callback will not be called, we just call
    // setBody and we are done.
    // 2) The callback was called docs.length times.
    // In this case all documents were created and we don’t need to call
    // tryCreate anymore. Just call setBody and we are done.
    function tryCreate(doc, callback) {
        var isAccepted = collection.createDocument(collectionLink, doc, { disableAutomaticIdGeneration : true}, callback);

    // If the request was accepted, callback will be called.
    // Otherwise report current count back to the client,
    // which will call the script again with remaining set of docs.
    if (!isAccepted) getContext().getResponse().setBody(count); 
}

// To replace the document, first issue a query to find it and then call replace.
function tryReplace(doc, callback) {
    var parsedDoc = JSON.parse(doc);
    retrieveDoc(parsedDoc, null, function(retrievedDocs){
        var isAccepted = collection.replaceDocument(retrievedDocs[0]._self, parsedDoc, callback);
        if (!isAccepted) getContext().getResponse().setBody(count);
    });
}

function retrieveDoc(doc, continuation, callback) {
    var query = "select * from root r where r.id = '" + doc.id + "'";
    var requestOptions = { continuation : continuation }; 
    var isAccepted = collection.queryDocuments(collectionLink, query, requestOptions, function(err, retrievedDocs, responseOptions) {
        if (err) throw err;

        if (retrievedDocs.length > 0) {
            callback(retrievedDocs);
        } else if (responseOptions.continuation) {
            retrieveDoc(doc, responseOptions.continuation, callback);            
        } else {
            throw "Error in retrieving document: " + doc.id;
        }
    });

    if (!isAccepted) getContext().getResponse().setBody(count); 
}

// This is called when collection.createDocument is done in order to
// process the result.
function callback(err, doc, options) {
    if (err) {
        // Replace the document if status code is 409 and upsert is enabled
        if(upsert && err.number == errorCodes.CONFLICT) {
            return tryReplace(docs[count], callback);
        } else {
            throw err;
        }
    }

       // One more document has been inserted, increment the count.
      count++;
      if (count >= docsLength) {
        // If we created all documents, we are done. Just set the response.
        getContext().getResponse().setBody(count); 
      } else {
        // Create next document.
        tryCreate(docs[count], callback);
      } 
}

由于存储过程有两个参数，我尝试了前面提到的两种方式，但都失败了。

---

更新代码：

print (param)
[[{'ACCOUNT': '1', 'id': '1', 'CASE_N': 'AB'}], True]

client.ExecuteStoredProcedure(sproc_link, [param],options=options)

仍然出现 400 错误：

400 子状态：400 {"code":"BadRequest","message":"Message: {\"Errors\":[\"执行函数时遇到异常。 Exception = Error: {\\"Errors\\":[\\"指定的输入之一是 无效\\"]}\r\n堆栈跟踪：错误：{\\"Errors\\":[\\"其中一个 指定的输入无效\\"]}\n

Answer 1

我已经修改了存储过程， 1.删​​除了upsert输入参数并使其硬编码， 2. 删除行 var parsedDoc = JSON.parse(doc); with var parsedDoc = doc;

现在程序运行正常。

Answer 2

我假设您遵循此 code sample 并且此代码工作正常。根据您的错误日志，我相信您的执行永远不会跳转到 upsert 分支。可能是您将参数错误地传递给 SP。

bulkimport SP 无需修改，而是为您的代码修改。

from pydocumentdb import document_client

endpoint = "https://***.documents.azure.com:443/";
primaryKey = "***";

client = document_client.DocumentClient(endpoint, {'masterKey': primaryKey})

options = {"partitionKey": "A"}

def create_cosmos_entity(jobid):
    return {
        'JobID': jobid,
        'id': jobid,
        "name": "A"
    }
bulkdos = []
bulkdos.append(create_cosmos_entity('1'))

param = []
param.append(bulkdos)
param.append(True)

sproc_link = "dbs/db/colls/coll/sprocs/bulkimport"
q = client.ExecuteStoredProcedure(sproc_link, [param], options)

你可以参考这个帖子：DocumentDB: bulkImport Stored Proc - Getting 400 Error on Array/JSON issue

“文档”必须是参数数组，否则， 程序执行器会将它们视为 过程，而不是单数组参数。