<pre name="code" class="python">#!/usr/bin/python # -*- coding: utf-8 -*- """ Created on 2016年11月5日 @author: hongyi 1、執行帶參數的ＳＱＬ時，請先用sql語句指定需要輸入的條件列表，然后再用tuple/list進行條件批配 2、在格式ＳＱＬ中不需要使用引號指定數據類型，系統會根據輸入參數自動識別 3、在輸入的值中不需要使用轉意函數，系統會自動處理 """ import MySQLdb import Config import sys from MySQLdb.cursors import DictCursor from DBUtils.PooledDB import PooledDB reload(sys) sys.setdefaultencoding('utf-8') """ Config是一些數據庫的配置文件 """ class Mysql(object): """ MYSQL數據庫對象，負責產生數據庫連接 , 此類中的連接采用連接池實現獲取連接對象：conn = Mysql.getConn() 釋放連接對象;conn.close()或del conn """ #連接池對象 __pool = None def __init__(self): #數據庫構造函數，從連接池中取出連接，并生成操作游標 try: self._conn = Mysql.__getConn() self._cursor = self._conn.cursor() except Exception, e: error = 'Connect failed! ERROR (%s): %s' %(e.args[0],e.args[1]) print error sys.exit() @staticmethod def __getConn(): """ @summary: 靜態方法，從連接池中取出連接 @return MySQLdb.connection """ if Mysql.__pool is None: __pool = PooledDB(creator=MySQLdb, mincached=1 , maxcached=20 , host=Config.DBHOST , port=Config.DBPORT ,user=Config.DBUSER , passwd=Config.DBPWD , db=Config.DBNAME,use_unicode=False,charset=Config.DBCHAR,cursorclass=DictCursor) return __pool.connection() #針對讀操作返回結果集 def _exeCute(self,sql=''): try: self._cursor.execute(sql) records = self._cursor.fetchall() return records except MySQLdb.Error,e: error = 'MySQL execute failed! ERROR (%s): %s' %(e.args[0],e.args[1]) print error #針對更新,刪除,事務等操作失敗時回滾 def _exeCuteCommit(self,sql='',arg=None): try: if arg is None: self._cursor.execute(sql) else: self._cursor.execute(sql,arg)self._conn.commit() except MySQLdb.Error,e: self._conn.rollback() error = 'MySQL execute failed! ERROR (%s): %s' %(e.args[0],e.args[1]) print error #sys.exit() #創建表 #tablename:表名稱,attr_dict:屬性鍵值對,constraint:主外鍵約束 #attr_dict:{'book_name':'varchar(200) NOT NULL'...} #constraint:PRIMARY KEY(`id`) def _createTable(self,table,attr_dict,constraint): sql = '' # sql_mid = '`row_id` bigint(11) NOT NULL AUTO_INCREMENT,' sql_mid = ''for attr,value in attr_dict.items(): sql_mid = sql_mid + '`'+attr + '`'+' '+ value+',' sql = sql + 'CREATE TABLE IF NOT EXISTS %s ('%table sql = sql + sql_mid sql = sql + constraint sql = sql + ') ENGINE=InnoDB DEFAULT CHARSET=utf8' print '_createTable:'+sql self._exeCuteCommit(sql) def insertOne(self,sql,value=None): """ @summary: 向數據表插入一條記錄 @param sql:要插入的ＳＱＬ格式 @param value:要插入的記錄數據tuple/list @return: insertId 受影響的行數 """ self._exeCuteCommit(sql,value)return self.__getInsertId() def _insert(self,table,attrs,value):"""@summary: 向數據表插入一條記錄 @param attrs = [] :要插入的屬性@param value = [] :要插入的數據值 """ #values_sql = ['%s' for v in attrs] attrs_sql = '('+','.join(attrs)+')' value_str = self._transferContent(value)values_sql = ' values('+ value_str +')' sql = 'insert into %s' %table sql = sql + attrs_sql + values_sql print '_insert:'+sql self._exeCuteCommit(sql) def _insertDic(self,table,attrs):"""@summary: 向數據表插入一條記錄 @param attrs = {"colNmae:value"} :要插入的屬性：數據值""" attrs_sql = '('+','.join(attrs.keys())+')' value_str = self._transferContent(attrs.values()) #','.join(attrs.values())values_sql = ' values('+ value_str +')' sql = 'insert into %s' %table sql = sql + attrs_sql + values_sql print '_insert:'+sql self._exeCuteCommit(sql) #將list轉為字符串def _transferContent(self, content):if content is None:return Noneelse:Strtmp = ""for col in content:if Strtmp == "":Strtmp = "\"" + col + "\""else:Strtmp += "," + "\"" + col + "\""return Strtmpdef _insertMany(self,table,attrs,values): """@summary: 向數據表插入多條數據 @param attrs = [id,name,...] :要插入的屬性@param values = [[1,'jack'],[2,'rose']] :要插入的數據值""" values_sql = ['%s' for v in attrs] attrs_sql = '('+','.join(attrs)+')' values_sql = ' values('+','.join(values_sql)+')' sql = 'insert into %s'%table sql = sql + attrs_sql + values_sql print '_insertMany:'+sql try: for i in range(0,len(values),20000): self._cursor.executemany(sql,values[i:i+20000]) self._conn.commit() except MySQLdb.Error,e: self._conn.rollback() error = '_insertMany executemany failed! ERROR (%s): %s' %(e.args[0],e.args[1]) print error sys.exit() def insertMany(self,sql,values=None): """ @summary: 向數據表插入多條記錄 @param sql:要插入的ＳＱＬ格式 @param values:要插入的記錄數據tuple(tuple)/list[list] @return: count 受影響的行數 """try: if values is None: count = self._cursor.executemany(sql) else: count = self._cursor.execute(sql,values)self._conn.commit() except MySQLdb.Error,e: self._conn.rollback() error = 'MySQL execute failed! ERROR (%s): %s' %(e.args[0],e.args[1]) print error sys.exit() return countdef _select(self,table,cond_dict='',order=''): """@summary: 執行條件查詢，并取出所有結果集@cond_dict:{'name':'xiaoming'...} @order:'order by id desc'@return: result ({"col":"val","":""},{})""" consql = ' ' if cond_dict!='': for k,v in cond_dict.items(): consql = consql+k+'='+v+' and' consql = consql + ' 1=1 ' sql = 'select * from %s where '%table sql = sql + consql + order print '_select:'+sql return self._exeCute(sql) def __getInsertId(self): """ 獲取當前連接最后一次插入操作生成的id,如果沒有則為０ """ self._cursor.execute("SELECT @@IDENTITY AS id") result = self._cursor.fetchall() return result[0]['id'] def __query(self,sql,param=None): if param is None: count = self._cursor.execute(sql) else: count = self._cursor.execute(sql,param) return count def getAll(self,sql,param=None): """ @summary: 執行查詢，并取出所有結果集 @param sql:查詢ＳＱＬ，如果有查詢條件，請只指定條件列表，并將條件值使用參數[param]傳遞進來 @param param: 可選參數，條件列表值（元組/列表） @return: result list(字典對象)/boolean 查詢到的結果集 """ if param is None: count = self._cursor.execute(sql) else: count = self._cursor.execute(sql,param) if count>0: result = self._cursor.fetchall() else: result = False return result def getOne(self,sql,param=None): """ @summary: 執行查詢，并取出第一條 @param sql:查詢ＳＱＬ，如果有查詢條件，請只指定條件列表，并將條件值使用參數[param]傳遞進來 @param param: 可選參數，條件列表值（元組/列表） @return: result list/boolean 查詢到的結果集 """ if param is None: count = self._cursor.execute(sql) else: count = self._cursor.execute(sql,param) if count>0: result = self._cursor.fetchone() else: result = False return result def getMany(self,sql,num,param=None): """ @summary: 執行查詢，并取出num條結果 @param sql:查詢ＳＱＬ，如果有查詢條件，請只指定條件列表，并將條件值使用參數[param]傳遞進來 @param num:取得的結果條數 @param param: 可選參數，條件列表值（元組/列表） @return: result list/boolean 查詢到的結果集 """ count = self.__query(sql,parm)if count>0: result = self._cursor.fetchmany(num) else: result = False return result def update(self,sql,param=None): """ @summary: 更新數據表記錄 @param sql: ＳＱＬ格式及條件，使用(%s,%s) @param param: 要更新的 值 tuple/list @return: count 受影響的行數 """ return self._exeCuteCommit(sql,param) def delete(self,sql,param=None): """ @summary: 刪除數據表記錄 @param sql: ＳＱＬ格式及條件，使用(%s,%s) @param param: 要刪除的條件 值 tuple/list @return: count 受影響的行數 """ return self._exeCuteCommit(sql,param) def begin(self): """ @summary: 開啟事務 """ self._conn.autocommit(0) def end(self,option='commit'): """ @summary: 結束事務 """ if option=='commit': self._conn.commit() else: self._conn.rollback() def dispose(self,isEnd=1): """ @summary: 釋放連接池資源 """ if isEnd==1: self.end('commit') else: self.end('rollback'); self._cursor.close() self._conn.close()

總結

以上是生活随笔為你收集整理的python操作数据库工具类的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。