測試連接到MySQL服務器

Question

什麼是最好的/正確的方式來測試連接到MySQL服務器..你可以舉例ping它..？我正在使用MySQLdb和python。

我想我的程序通過以下方式

....connect to MySQL server 
database = MySQLdb.connect(host="127.0.0.1 etc... 

While true: 
    **... Check to see if connection is still alive if not reconnect** 

    ... send data to MySQL... 
    time.sleep(30) 

Answer 1

結構化這是我用過的東西。

import MySQLdb 

try: 
    import MySQLdb.converters 
except ImportError: 
    _connarg('conv') 

def connect(host='ronak.local', user='my_dev_1', passwd='my_dev_1', db='my_dev1', port=3306): 

    try: 
     orig_conv = MySQLdb.converters.conversions 
     conv_iter = iter(orig_conv) 
     convert = dict(zip(conv_iter, [str,] * len(orig_conv.keys()))) 
     print "Connecting host=%s user=%s db=%s port=%d" % (host, user, db, port) 
     conn = MySQLdb.connect(host, user, passwd, db, port, conv=convert) 
    except MySQLdb.Error, e: 
     print "Error connecting %d: %s" % (e.args[0], e.args[1]) 
    return conn 


def parse_data_and_description(cursor, data, rs_id): 

    res = [] 
    cols = [d[0] for d in cursor.description] 

    for i in data: 
     res.append(OrderedDict(zip(cols, i))) 
    return res 
    rs_id=0; 

def get_multiple_result_sets(): 
    conn = connect() 
    cursor = conn.cursor() 
    final_list = [] 
    try: 
     conn.autocommit(True) 
     cursor.execute ("CALL %s%s" % (sp, args)) 
     while True: 
      rs_id+=1 

      data = cursor.fetchall() 
      listout = parse_data_and_description(cursor, data, rs_id) 
      print listout 
      if cursor.nextset()==None: 
       # This means no more recordsets available 
       break 
      print "\n" 
      # Consolidate all the cursors in a single list 
      final_list.append(listout) 
     print final_list 
    except MySQLdb.Error, e: 
     # Lets rollback the transaction in case of an exception 
     conn.rollback() 
     print "Transaction aborted: %d: %s" % (e.args[0], e.args[1]) 
     cursor.close() 
     conn.close() 
    else: 
     # Commit the transaction in case of no failures/exceptions 
     conn.commit() 
     print "Transaction succeeded" 
     cursor.close() 
     conn.close()