上一篇中我们分析了数据库测试的几种应用类型，不知道其他公司还有什么数据库应用测试类型，可以分享出来，大家一起探讨下测试方案。具体的案列代码我就不在这里贴出来了，这个和具体业务高度相关。这里贴两个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> |
