上一篇中我们分析了数据库测试的几种应用类型，不知道其他公司还有什么数据库应用测试类型，可以分享出来，大家一起探讨下测试方案。具体的案列代码我不在这里贴出来了，这个和具体业务高度相关。这里贴两个MySQL， vertical数据库的连接和SQL 操作的类。大家以后应用可以直接用了。
　　MySQL 数据库连接与SQL 操作
import MySQLdb，time，traceback，logging
class DBOperateAction:
def __init__(self，dbhost，dbaccount，dbpasswd，dbname，port=3306，charset="utf8"):
self.dbhost=dbhost
self.dbaccount=dbaccount
self.dbpasswd=dbpasswd
self.dbname=dbname
self.charset=charset
self.port=port
self.db_conn=""
self.db_cursor=""
def connect(self):
try:
self.db_conn=MySQLdb.connect(host=self.dbhost，user=self.dbaccount，passwd=self.dbpasswd，
db=self.dbname，port=self.port，charset=self.charset)
self.db_cursor=self.db_conn.cursor()
return True
except MySQLdb.OperationalError:
logging.error("Connect to "+self.dbhost+" Failed")
logging.exception("exception message:")
return False
def re_connect(self):
logging.error("connect to mysql server failed， reconnect")
try:
self.db_conn=MySQLdb.connect(host=self.dbhost，user=self.dbaccount，passwd=self.dbpasswd，
db=self.dbname，port=self.port，charset="utf8")
self.db_cursor=self.db_conn.cursor()
return True
except MySQLdb.OperationalError:
logging.error("Reconnect MySQL failed")
return False
def get_all_result(self，sql):
try:
logging.info("Execute sql: "+sql)
self.db_cursor.execute(sql)
self.db_conn.commit()
result=self.db_cursor.fetchall()
return result
except MySQLdb.OperationalError:
logging.exception("Execure SQL except"+sql)
for i in range(0，3):
time.sleep(5)
if self.re_connect():
logging.info("reconnect database successfully")
return False
def get_one_result(self，sql):
try:
logging.info("Execute sql: "+sql)
self.db_cursor.execute(sql)
self.db_conn.commit()
result=self.db_cursor.fetchone()
return result
except MySQLdb.OperationalError:
traceback.print_exc()
for i in range(0，3):
time.sleep(5)
if self.re_connect():
break
def close_connection(self):
self.db_conn.close()
　　Vertica 数据库连接与SQL 操作. 需要注意的是， Vertical里面schema 和database 不个东西。在mysql里面schema和database 是一回事。
import vertica_python
class VerticaDBOperateAction:
def __init__(self，dbhost，dbaccount，dbpasswd，dbname，port=5433):
#self._conn_info=conn_info
self.connection=""
self.cursorcur =""
self.dbhost=dbhost
self.dbaccount=dbaccount
self.dbpasswd=dbpasswd
self.dbname=dbname
self.port=port
def connect(self):
try:
self.connection = vertica_python.connect(host=self.dbhost，user=self.dbaccount，password=self.dbpasswd，
database=self.dbname，port=self.port)
self.cursorcur=self.connection.cursor()
return True
except:
logging.exception("Connect vertica database error with address:"+self.dbhost)
logging.exception("exception message:")
return False
def get_all_result(self，sql):
'''''
:return: # [ [1， 'something']， [2， 'something_else'] ]
'''
self.cursorcur.execute(sql)
return self.cursorcur.fetchall()
def close_connection(self):
self.connection.close()
<pre code_snippet_id="2215959" snippet_file_name="blog_20170219_2_4304522"></pre>
<pre></pre>