err in [
        ER_DUP_ENTRY,
        ER_DUP_UNIQUE,
        ER_PRIMARY_CANT_HAVE_NULL]:
        return IntegrityError

    if merr in [ER_WARNING_NOT_COMPLETE_ROLLBACK]:
        return NotSupportedError

    if merr < 1000:
        return InternalError
    else:
        return OperationalError

# ===========================================================================
#                           Authentication
# ===========================================================================
#
# switching to 4.1 password/handshake.
#
# mysql-4.1.13a/libmysql/password.c
#
#  SERVER:  public_seed=create_random_string()
#           send(public_seed)
#
#  CLIENT:  recv(public_seed)
#           hash_stage1=sha1("password")
#           hash_stage2=sha1(hash_stage1)
#           reply=xor(hash_stage1, sha1(public_seed,hash_stage2)
#           send(reply)
#
#  SERVER:  recv(reply)
#           hash_stage1=xor(reply, sha1(public_seed,hash_stage2))
#           candidate_hash2=sha1(hash_stage1)
#           check(candidate_hash2==hash_stage2)

def scramble (message, password):
    hash_stage1 = sha.new(password).digest()
    hash_stage2 = sha.new(hash_stage1).digest()
    hash_stage2 = sha.new(message + hash_stage2).digest()

    reply = ''
    while hash_stage1 and hash_stage2:
        reply = reply + chr(ord(hash_stage1[0]) ^ ord(hash_stage2[0]))
        hash_stage1 = hash_stage1[1:]
        hash_stage2 = hash_stage2[1:]

    return reply
# ===========================================================================
#                           Packet Protocol
# ===========================================================================

def unpacket (p):
    # 3-byte length, one-byte packet number, followed by packet data
    a,b,c,s = map (ord, p[:4])
    l = a | (b << 8) | (c << 16)
    # s is a sequence number

    return l, s

def packet(data, s = 0):
    return struct.pack('<i', len(data) | s << 24) + data

def n_byte_num (data, n, pos=0):
    result = 0
    for i in range(n):
        result = result | (ord(data[pos+i])<<(8*i))

    return result

def decode_length (data):
    n = ord(data[0])

    if n < 251:
        return n, 1
    elif n == 251:
        return 0, 1
    elif n == 252:
        return n_byte_num (data, 2, 1), 3
    elif n == 253:
        # documentation says 32 bits, but wire decode shows 24 bits.
        # (there is enough padding on the wire for 32, but lets stick
        #  to 24...) Rollover to 254 happens at 24 bit boundry.
        return n_byte_num (data, 3, 1), 4
    elif n == 254:
        return n_byte_num (data, 8, 1), 9
    else:
        raise DataError, ('unexpected field length', n)

# used to generate the dumps below
def dump_hex (s):
    r1 = []
    r2 = []

    for ch in s:
        r1.append (' %02x' % ord(ch))
        if (ch in string.letters) or (ch in string.digits):
            r2.append ('  %c' % ch)
        else:
            r2.append ('   ')

    return string.join (r1, ''), string.join (r2, '')
# ===========================================================================
# generic utils
# ===========================================================================
def is_disconnect(reason):
    lower = string.lower(repr(reason.args))
    if string.find(lower, "lost connection")!= -1:
        return 1

    if string.find(lower, "no connection")!= -1:
        return 1

    if string.find(lower, "server has gone away")!= -1:
        return 1

    return 0

def handle_error_sanely(cursor, errclass, errargs):
    raise

class connection(object):
    server_capabilities = 0

    def __init__ (self, **kwargs):
        self.username = kwargs.get('user',   '')
        self.password = kwargs.get('passwd', '')

        self.addresses = [(
            socket.AF_INET,
            (kwargs.get('host', '127.0.0.1'), kwargs.get('port', 3306)))]

        if kwargs.get('host') == 'localhost':
            self.addresses.append((socket.AF_UNIX, '/tmp/mysql.sock'))

        self._database = kwargs.get('db', '')
        self._init_cmd = kwargs.get('init_command', None)

        self._debug = kwargs.get('debug', 0) or (coro.current_id() < 0)

        self._connect_timeout = kwargs.get('connect_timeout', None)
        self._timeout         = kwargs.get('timeout', None)

        self._connected   = 0
        self._server_info = ''
        self._recv_buffer = ''

        self._rbuf = ''
        self._rlen = 0
        self._roff = 0
        self._rpac = []

        self._latest_fields = []
        self._latest_rows   = []
        self._nfields       = 0
        self._affected      = 0
        self._insert_id     = 0
        self._warning_count = 0
        self._message       = None

        self.socket = None
        self._lock  = 0

        self._reconnect_cmds = 0
        self._reconnect_exec = 0
        self._reconnect_maxr = MAX_RECONNECT_RETRY

        if not self._debug:
            self._timer_cond = coro.coroutine_cond()
            self._lock_cond  = coro.coroutine_cond()

        self.converter = {}
        self.errorhandler = handle_error_sanely

        from MySQLdb import converters
        from weakref import proxy

        self.charset = self.character_set_name().split('_')[0]

        self.converter = converters.conversions.copy()
        self.converter[types.StringType]  = self.string_literal
        self.converter[types.UnicodeType] = self.unicode_literal
        #
        # preconverter set by upper layer, blech, use it for string types.
        #
        self._preconv = kwargs.get('conv', None)
        #
        # check switch for C implementation.
        #
        if kwargs.get('fast', False) and mysqlfuncs:
            self.read_packet = self._fast_read_packet
            self.cmd_query   = self._fast_cmd_query

        return None

    def __del__(self):
        self._close()

    def handle_error_sanely(self, self2, errclass, errargs):
        raise

    def sleep(self, *args, **kwargs):
        if self._debug:
            return apply(time.sleep, args, kwargs)
        else:
            return apply(self._timer_cond.wait, args, kwargs)

    def make_socket(self, *args, **kwargs):
        if self._debug:
            return apply(socket.socket, args, kwargs)
        else:
            return apply(coro.make_socket, args, kwargs)

    def lock_wait(self, *args, **kwargs):
        if not self._debug:
            return apply(self._lock_cond.wait, args, kwargs)

    def lock_wake(self, *args, **kwargs):
        if not self._debug:
            return apply(self._lock_cond.wake_one, args, kwargs)

    def lock_waiting(self):
        if not self._debug:
            return len(self._lock_cond)
        else:
            return 0

    def _lock_connection(self):

        while self._lock:
            self.lock_wait()

        self._lock = 1
        return None

    def _unlock_connection(self):
