<s> """<STR_LIT>""" <EOL> import collections <EOL> import datetime <EOL> import struct <EOL> import bson <EOL> from bson . codec_options import CodecOptions <EOL> from bson . py3compat import itervalues , string_type , iteritems , u <EOL> from bson . son import SON <EOL> from pymongo import ASCENDING <EOL> from pymongo . errors import ( CursorNotFound , <EOL> DuplicateKeyError , <EOL> ExecutionTimeout , <EOL> NotMasterError , <EOL> OperationFailure , <EOL> WriteError , <EOL> WriteConcernError , <EOL> WTimeoutError ) <EOL> from pymongo . message import _Query , _convert_exception <EOL> from pymongo . read_concern import DEFAULT_READ_CONCERN <EOL> _UUNDER = u ( "<STR_LIT:_>" ) <EOL> def _gen_index_name ( keys ) : <EOL> """<STR_LIT>""" <EOL> return _UUNDER . join ( [ "<STR_LIT>" % item for item in keys ] ) <EOL> def _index_list ( key_or_list , direction = None ) : <EOL> """<STR_LIT>""" <EOL> if direction is not None : <EOL> return [ ( key_or_list , direction ) ] <EOL> else : <EOL> if isinstance ( key_or_list , string_type ) : <EOL> return [ ( key_or_list , ASCENDING ) ] <EOL> elif not isinstance ( key_or_list , ( list , tuple ) ) : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return key_or_list <EOL> def _index_document ( index_list ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( index_list , collections . Mapping ) : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % list ( iteritems ( index_list ) ) ) <EOL> elif not isinstance ( index_list , ( list , tuple ) ) : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" + repr ( index_list ) ) <EOL> if not len ( index_list ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> index = SON ( ) <EOL> for ( key , value ) in index_list : <EOL> if not isinstance ( key , string_type ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if not isinstance ( value , ( string_type , int , collections . Mapping ) ) : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> index [ key ] = value <EOL> return index <EOL> def _unpack_response ( response , cursor_id = None , codec_options = CodecOptions ( ) ) : <EOL> """<STR_LIT>""" <EOL> response_flag = struct . unpack ( "<STR_LIT>" , response [ : <NUM_LIT:4> ] ) [ <NUM_LIT:0> ] <EOL> if response_flag & <NUM_LIT:1> : <EOL> assert cursor_id is not None <EOL> msg = "<STR_LIT>" % ( cursor_id , ) <EOL> errobj = { "<STR_LIT>" : <NUM_LIT:0> , "<STR_LIT>" : msg , "<STR_LIT:code>" : <NUM_LIT> } <EOL> raise CursorNotFound ( msg , <NUM_LIT> , errobj ) <EOL> elif response_flag & <NUM_LIT:2> : <EOL> error_object = bson . BSON ( response [ <NUM_LIT:20> : ] ) . decode ( ) <EOL> error_object . setdefault ( "<STR_LIT>" , <NUM_LIT:0> ) <EOL> if error_object [ "<STR_LIT>" ] . startswith ( "<STR_LIT>" ) : <EOL> raise NotMasterError ( error_object [ "<STR_LIT>" ] , error_object ) <EOL> elif error_object . get ( "<STR_LIT:code>" ) == <NUM_LIT:50> : <EOL> raise ExecutionTimeout ( error_object . get ( "<STR_LIT>" ) , <EOL> error_object . get ( "<STR_LIT:code>" ) , <EOL> error_object ) <EOL> raise OperationFailure ( "<STR_LIT>" % <EOL> error_object . get ( "<STR_LIT>" ) , <EOL> error_object . get ( "<STR_LIT:code>" ) , <EOL> error_object ) <EOL> result = { "<STR_LIT>" : struct . unpack ( "<STR_LIT>" , response [ <NUM_LIT:4> : <NUM_LIT:12> ] ) [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" : struct . unpack ( "<STR_LIT>" , response [ <NUM_LIT:12> : <NUM_LIT:16> ] ) [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" : struct . unpack ( "<STR_LIT>" , response [ <NUM_LIT:16> : <NUM_LIT:20> ] ) [ <NUM_LIT:0> ] , <EOL> "<STR_LIT:data>" : bson . decode_all ( response [ <NUM_LIT:20> : ] , codec_options ) } <EOL> assert len ( result [ "<STR_LIT:data>" ] ) == result [ "<STR_LIT>" ] <EOL> return result <EOL> def _check_command_response ( response , msg = None , allowable_errors = None ) : <EOL> """<STR_LIT>""" <EOL> if "<STR_LIT>" not in response : <EOL> raise OperationFailure ( response . get ( "<STR_LIT>" ) , <EOL> response . get ( "<STR_LIT:code>" ) , <EOL> response ) <EOL> if response . get ( "<STR_LIT>" , False ) : <EOL> raise WTimeoutError ( response . get ( "<STR_LIT>" , response . get ( "<STR_LIT>" ) ) , <EOL> response . get ( "<STR_LIT:code>" ) , <EOL> response ) <EOL> if not response [ "<STR_LIT>" ] : <EOL> details = response <EOL> if "<STR_LIT>" in response : <EOL> for shard in itervalues ( response [ "<STR_LIT>" ] ) : <EOL> if shard . get ( "<STR_LIT>" ) and not shard . get ( "<STR_LIT>" ) : <EOL> details = shard <EOL> break <EOL> errmsg = details [ "<STR_LIT>" ] <EOL> if allowable_errors is None or errmsg not in allowable_errors : <EOL> if ( errmsg . startswith ( "<STR_LIT>" ) <EOL> or errmsg . startswith ( "<STR_LIT>" ) ) : <EOL> raise NotMasterError ( errmsg , response ) <EOL> if errmsg == "<STR_LIT>" : <EOL> errmsg = ( "<STR_LIT>" % <EOL> details . get ( "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> raise OperationFailure ( errmsg , <EOL> details . get ( "<STR_LIT>" ) , <EOL> response ) <EOL> code = details . get ( "<STR_LIT:code>" ) <EOL> if code in ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) : <EOL> raise DuplicateKeyError ( errmsg , code , response ) <EOL> elif code == <NUM_LIT:50> : <EOL> raise ExecutionTimeout ( errmsg , code , response ) <EOL> elif code == <NUM_LIT> : <EOL> raise CursorNotFound ( errmsg , code , response ) <EOL> msg = msg or "<STR_LIT:%s>" <EOL> raise OperationFailure ( msg % errmsg , code , response ) <EOL> def _check_gle_response ( response ) : <EOL> """<STR_LIT>""" <EOL> response = _unpack_response ( response ) <EOL> assert response [ "<STR_LIT>" ] == <NUM_LIT:1> <EOL> result = response [ "<STR_LIT:data>" ] [ <NUM_LIT:0> ] <EOL> _check_command_response ( result ) <EOL> if result . get ( "<STR_LIT>" , False ) : <EOL> raise WTimeoutError ( result . get ( "<STR_LIT>" , result . get ( "<STR_LIT>" ) ) , <EOL> result . get ( "<STR_LIT:code>" ) , <EOL> result ) <EOL> error_msg = result . get ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if error_msg is None : <EOL> return result <EOL> if error_msg . startswith ( "<STR_LIT>" ) : <EOL> raise NotMasterError ( error_msg , result ) <EOL> details = result <EOL> if "<STR_LIT>" in result : <EOL> for errobj in result [ "<STR_LIT>" ] : <EOL> if errobj . get ( "<STR_LIT>" ) == error_msg : <EOL> details = errobj <EOL> break <EOL> code = details . get ( "<STR_LIT:code>" ) <EOL> if code in ( <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) : <EOL> raise DuplicateKeyError ( details [ "<STR_LIT>" ] , code , result ) <EOL> raise OperationFailure ( details [ "<STR_LIT>" ] , code , result ) <EOL> def _first_batch ( sock_info , db , coll , query , ntoreturn , <EOL> slave_ok , codec_options , read_preference , cmd , listeners ) : <EOL> """<STR_LIT>""" <EOL> query = _Query ( <EOL> <NUM_LIT:0> , db , coll , <NUM_LIT:0> , query , None , <EOL> codec_options , read_preference , ntoreturn , <NUM_LIT:0> , DEFAULT_READ_CONCERN ) <EOL> name = next ( iter ( cmd ) ) <EOL> duration = None <EOL> publish = listeners . enabled_for_commands <EOL> if publish : <EOL> start = datetime . datetime . now ( ) <EOL> request_id , msg , max_doc_size = query . get_message ( slave_ok , <EOL> sock_info . is_mongos ) <EOL> if publish : <EOL> encoding_duration = datetime . datetime . now ( ) - start <EOL> listeners . publish_command_start ( <EOL> cmd , db , request_id , sock_info . address ) <EOL> start = datetime . datetime . now ( ) <EOL> sock_info . send_message ( msg , max_doc_size ) <EOL> response = sock_info . receive_message ( <NUM_LIT:1> , request_id ) <EOL> try : <EOL> result = _unpack_response ( response , None , codec_options ) <EOL> except Exception as exc : <EOL> if publish : <EOL> duration = ( datetime . datetime . now ( ) - start ) + encoding_duration <EOL> if isinstance ( exc , ( NotMasterError , OperationFailure ) ) : <EOL> failure = exc . details <EOL> else : <EOL> failure = _convert_exception ( exc ) <EOL> listeners . publish_command_failure ( <EOL> duration , failure , name , request_id , sock_info . address ) <EOL> raise <EOL> if publish : <EOL> duration = ( datetime . datetime . now ( ) - start ) + encoding_duration <EOL> listeners . publish_command_success ( <EOL> duration , result , name , request_id , sock_info . address ) <EOL> return result <EOL> def _check_write_command_response ( results ) : <EOL> """<STR_LIT>""" <EOL> errors = [ res for res in results <EOL> if "<STR_LIT>" in res [ <NUM_LIT:1> ] or "<STR_LIT>" in res [ <NUM_LIT:1> ] ] <EOL> if errors : <EOL> offset , result = errors [ - <NUM_LIT:1> ] <EOL> write_errors = result . get ( "<STR_LIT>" ) <EOL> if write_errors : <EOL> error = write_errors [ - <NUM_LIT:1> ] <EOL> error [ "<STR_LIT:index>" ] += offset <EOL> if error . get ( "<STR_LIT:code>" ) == <NUM_LIT> : <EOL> raise DuplicateKeyError ( error . get ( "<STR_LIT>" ) , <NUM_LIT> , error ) <EOL> raise WriteError ( error . get ( "<STR_LIT>" ) , error . get ( "<STR_LIT:code>" ) , error ) <EOL> else : <EOL> error = result [ "<STR_LIT>" ] <EOL> if "<STR_LIT>" in error and error [ "<STR_LIT>" ] . get ( '<STR_LIT>' ) : <EOL> raise WTimeoutError ( <EOL> error . get ( "<STR_LIT>" ) , error . get ( "<STR_LIT:code>" ) , error ) <EOL> raise WriteConcernError ( <EOL> error . get ( "<STR_LIT>" ) , error . get ( "<STR_LIT:code>" ) , error ) <EOL> def _fields_list_to_dict ( fields , option_name ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( fields , collections . Mapping ) : <EOL> return fields <EOL> if isinstance ( fields , collections . Sequence ) : <EOL> if not all ( isinstance ( field , string_type ) for field in fields ) : <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( option_name , <EOL> string_type . __name__ ) ) <EOL> return dict . fromkeys ( fields , <NUM_LIT:1> ) <EOL> raise TypeError ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( option_name , ) ) </s>
