Python pymongo.errors 模块，BulkWriteError() 实例源码

def writeImportsTree(self, imports):
        command = {"$inc": {"count": 1}}
        bulk = self.import_coll.initialize_unordered_bulk_op()
        execute_bool = False
        for i in imports:
            dll_name = i["lib"]
            funcs = i["functions"]
            for imp_name in funcs:
                execute_bool = True
                bulk.find({"function_name": imp_name.lower(),
                           "dll_name": dll_name.lower()}).upsert().update(command)
                # print("**** Error Imports Tree ****")
                # err=str(traceback.format_exc())
                # print(err)
                # return -1
        try:
            if(execute_bool):
                bulk.execute({'w': 0})
        except BulkWriteError:
            logging.exception("MetaController(): " +
                              str("**** Error Imports Tree ****"))
            # err=str(traceback.format_exc())
            # print(err)
            return -1
        return 0

def mongo_do_bulk_insert(target_collection, documents_to_insert):
    assert isinstance(target_collection, collection.Collection)
    assert isinstance(documents_to_insert, cursor.Cursor)

    print("Doing bulk insert of [%s] documents into destination [%s]" % (
        documents_to_insert.count(), target_collection.database.name + "." + target_collection.name))

    try:
        result = target_collection.insert_many(documents_to_insert)
    except BulkWriteError as bwe:
        pprint(bwe.details)
        exit()

    inserted_count = len(result.inserted_ids)

    if inserted_count == documents_to_insert.count():
        print("Successfully inserted all [%d] documents." % inserted_count)
    elif inserted_count < documents_to_insert.count():
        print("Not all insertions succeeded.  Inserted [%d] out of [%d] documents." % (
            inserted_count, documents_to_insert.count()))
    else:
        print("ERROR: Inserted [%d] documents, which is more than documents_to_insert.count() [%d]." % (
            inserted_count, documents_to_insert.count()))
        exit()

def execute_command(self, connection: 'aiomongo.Connection', generator: Iterable[_Run],
                              write_concern: WriteConcern) -> dict:
        """Execute using write commands.
        """
        # nModified is only reported for write commands, not legacy ops.
        full_result = {
            'writeErrors': [],
            'writeConcernErrors': [],
            'nInserted': 0,
            'nUpserted': 0,
            'nMatched': 0,
            'nModified': 0,
            'nRemoved': 0,
            'upserted': [],
        }

        for run in generator:
            cmd = SON([(_COMMANDS[run.op_type], self.collection.name),
                       ('ordered', self.ordered)])
            if write_concern.document:
                cmd['writeConcern'] = write_concern.document
            if self.bypass_doc_val and connection.max_wire_version >= 4:
                cmd['bypassDocumentValidation'] = True

            results = await self._do_batched_write_command(
                self.namespace, run.op_type, cmd,
                run.ops, True, self.collection.codec_options, connection)

            _merge_command(run, full_result, results)
            # We're supposed to continue if errors are
            # at the write concern level (e.g. wtimeout)
            if self.ordered and full_result['writeErrors']:
                break

        if full_result['writeErrors'] or full_result['writeConcernErrors']:
            if full_result['writeErrors']:
                full_result['writeErrors'].sort(
                    key=lambda error: error['index'])
            raise BulkWriteError(full_result)
        return full_result

def save_stock_list(self, stocks):
        if len(stocks) > 0:
            try:
                stocklist_collection = self.db[self.STOCK_LIST]
                stocklist_collection.insert_many(stocks, ordered=False)
            except (DuplicateKeyError, BulkWriteError) as err:
                logger.error("save_stock_list: %i %s", len(stocks), err)

def save_stock_historical_data(self, quote, stock_historical_data_array):
        if len(stock_historical_data_array) > 0:
            try:
                collection_name = quote + self.HISTORICAL_DATA_SUFIX
                self.create_historical_collection_if_not_exists(collection_name)
                stock_historical_data_collection = self.db[collection_name]
                stock_historical_data_collection.insert_many(stock_historical_data_array, ordered=False)
            except (DuplicateKeyError, BulkWriteError) as err:
                logger.error("save_stock_historical_data: %s %i %s", quote, len(stock_historical_data_array), err)

def update(self, statement):
        from pymongo import UpdateOne
        from pymongo.errors import BulkWriteError

        data = statement.serialize()

        operations = []

        update_operation = UpdateOne(
            {'text': statement.text},
            {'$set': data},
            upsert=True
        )
        operations.append(update_operation)

        # Make sure that an entry for each response is saved
        for response_dict in data.get('in_response_to', []):
            response_text = response_dict.get('text')

            # $setOnInsert does nothing if the document is not created
            update_operation = UpdateOne(
                {'text': response_text},
                {'$set': response_dict},
                upsert=True
            )
            operations.append(update_operation)

        try:
            self.statements.bulk_write(operations, ordered=False)
        except BulkWriteError as bwe:
            # Log the details of a bulk write error
            self.logger.error(str(bwe.details))

        return statement