(collection_name, docs, check_keys,
                       safe, last_error_args, continue_on_error, opts,
                       ctx)
Insert `docs` using multiple batches.
Insert `docs` using multiple batches.
Insert docs using multiple batches .
def _do_batched_insert(collection_name, docs, check_keys,
                       safe, last_error_args, continue_on_error, opts,
                       ctx):
    """Insert `docs` using multiple batches.
    """
    def _insert_message(insert_message, send_safe):
        """Build the insert message with header and GLE.
        """
        request_id, final_message = __pack_message(2002, insert_message)
        if send_safe:
            request_id, error_message, _ = __last_error(collection_name,
                                                        last_error_args)
            final_message += error_message
        return request_id, final_message

    send_safe = safe or not continue_on_error
    last_error = None
    data = StringIO()
    data.write(struct.pack("<i", int(continue_on_error)))
    data.write(bson._make_c_string(collection_name))
    message_length = begin_loc = data.tell()
    has_docs = False
    to_send = []
    for doc in docs:
        encoded = bson.BSON.encode(doc, check_keys, opts)
        encoded_length = len(encoded)
        too_large = (encoded_length > ctx.max_bson_size)

        message_length += encoded_length
        if message_length < ctx.max_message_size and not too_large:
            data.write(encoded)
            to_send.append(doc)
            has_docs = True
            continue

        if has_docs:
            # We have enough data, send this message.
            try:
                request_id, msg = _insert_message(data.getvalue(), send_safe)
                ctx.legacy_write(request_id, msg, 0, send_safe, to_send)
            # Exception type could be OperationFailure or a subtype
            # (e.g. DuplicateKeyError)
            except OperationFailure as exc:
                # Like it says, continue on error...
                if continue_on_error:
                    # Store exception details to re-raise after the final batch.
                    last_error = exc
                # With unacknowledged writes just return at the first error.
                elif not safe:
                    return
                # With acknowledged writes raise immediately.
                else:
                    raise

        if too_large:
            _raise_document_too_large(
                "insert", encoded_length, ctx.max_bson_size)

        message_length = begin_loc + encoded_length
        data.seek(begin_loc)
        data.truncate()
        data.write(encoded)
        to_send = [doc]

    if not has_docs:
        raise InvalidOperation("cannot do an empty bulk insert")

    request_id, msg = _insert_message(data.getvalue(), safe)
    ctx.legacy_write(request_id, msg, 0, safe, to_send)

    # Re-raise any exception stored due to continue_on_error
    if last_error is not None:
        raise last_error
def _do_batched_insert ( collection_name , docs , check_keys , safe , last_error_args , continue_on_error , opts , ctx ) : def _insert_message ( insert_message , send_safe ) : """Build the insert message with header and GLE.
        """ request_id , final_message = __pack_message ( 2002 , insert_message ) if send_safe : request_id , error_message , _ = __last_error ( collection_name , last_error_args ) final_message += error_message return request_id , final_message send_safe = safe or not continue_on_error last_error = None data = StringIO ( ) data . write ( struct . pack ( "<i" , int ( continue_on_error ) ) ) data . write ( bson . _make_c_string ( collection_name ) ) message_length = begin_loc = data . tell ( ) has_docs = False to_send = [ ] for doc in docs : encoded = bson . BSON . encode ( doc , check_keys , opts ) encoded_length = len ( encoded ) too_large = ( encoded_length > ctx . max_bson_size ) message_length += encoded_length if message_length < ctx . max_message_size and not too_large : data . write ( encoded ) to_send . append ( doc ) has_docs = True continue if has_docs : # We have enough data, send this message. try : request_id , msg = _insert_message ( data . getvalue ( ) , send_safe ) ctx . legacy_write ( request_id , msg , 0 , send_safe , to_send ) # Exception type could be OperationFailure or a subtype # (e.g. DuplicateKeyError) except OperationFailure as exc : # Like it says, continue on error... if continue_on_error : # Store exception details to re-raise after the final batch. last_error = exc # With unacknowledged writes just return at the first error. elif not safe : return # With acknowledged writes raise immediately. else : raise if too_large : _raise_document_too_large ( "insert" , encoded_length , ctx . max_bson_size ) message_length = begin_loc + encoded_length data . seek ( begin_loc ) data . truncate ( ) data . write ( encoded ) to_send = [ doc ] if not has_docs : raise InvalidOperation ( "cannot do an empty bulk insert" ) request_id , msg = _insert_message ( data . getvalue ( ) , safe ) ctx . legacy_write ( request_id , msg , 0 , safe , to_send ) # Re-raise any exception stored due to continue_on_error if last_error is not None : raise last_error