Python封装的访问MySQL数据库的类及DEMO

  1 # Filename：mysql_class.py
  2 # Author:Rain.Zen; Date: 2014-04-15
  3 
  4 import MySQLdb
  5 
  6 class MyDb:
  7 
  8     '''初始化[类似于构造函数]'''
  9     def __init__(self, host, user, password, charset = 'utf8'):
 10         self.host = host
 11         self.user = user
 12         self.password = password
 13         self.charset = charset
 14         try:
 15             self.conn = MySQLdb.connect(host = self.host, user = self.user, passwd = self.password, charset = self.charset)
 16             self.cur = self.conn.cursor()
 17         except MySQLdb.Error as e:
 18             print('MySQL Error %d: %s' % (e.args[0], e.args[1]))
 19 
 20     '''组合字符串'''
 21     def joinData(self, data, char = ','):
 22         return char.join(data)
 23 
 24     '''解析条件语句，参数类型[{dict}, 'AND|OR'] | {dict}'''
 25     def parseMap(self, condition) :
 26         if isinstance(condition, list) :
 27             dictMap = condition[0]
 28             logic = ' '+ condition[1] +' '
 29         else :
 30             dictMap = condition
 31             logic = ' AND '
 32 
 33         if isinstance(dictMap, dict) :
 34             for k, v in dictMap.items():
 35                 dictMap[k] = "`"+ k +"` = '"+ str(v) +"'"
 36             return self.joinData(dictMap.values(), logic)
 37         else :
 38             return '1 = 1'
 39         
 40     '''选择数据表'''
 41     def selectDb(self, db):
 42       try :
 43           self.conn.select_db(db)
 44       except MySQLdb.Error as e:
 45           print('MySQL Error %d: %s' % (e.args[0], e.args[1]))
 46 
 47     '''执行SQL语句'''
 48     def query(self, sql):
 49         try :
 50             return self.cur.execute(sql)
 51         except MySQLdb.Error as e:
 52             print 'MySQL Error: %s 
SQL: %s' % (e, sql)
 53 
 54     '''添加一条信息'''
 55     def addOne(self, table, dictData):
 56         for field, value in dictData.items():
 57             dictData[field] = "'"+ str(value) +"'"
 58         self.query('INSERT INTO ' + table + "("+ self.joinData(dictData.keys()) +') VALUES('+ self.joinData(dictData.values()) +')')
 59         return self.cur.lastrowid
 60 
 61     '''修改一条信息[根据条件]'''
 62     def saveOne(self, table, dictData, condition):
 63         for key, val in dictData.items():
 64             dictData[key] = "`"+ key +"` = '"+ str(val) +"'"
 65         save_sql = 'UPDATE ' + table + ' SET ' + self.joinData(dictData.values()) + ' WHERE ' + self.parseMap(condition)
 66         return self.query(save_sql)
 67 
 68     '''删除信息[根据条件]'''
 69     def deleteOne(self, table, condition):
 70         return self.query('DELETE FROM ' + table + ' WHERE ' + self.parseMap(condition))
 71 
 72     '''读取单条信息[根据条件]'''
 73     def fetchOne(self, tabel, condition, fields = '*', dataType = 0):
 74         field = isinstance(fields, list) and self.joinData(fields) or fields
 75         self.query('SELECT '+ field +' FROM ' + tabel + ' WHERE ' + self.parseMap(condition))
 76         tupleData = self.cur.fetchone()
 77         if dataType == 0 or fields == '*' :
 78             return tupleData
 79         else :
 80             dictData = {}
 81             n = 0
 82             for k in fields:
 83                 dictData[k] = tupleData[n]
 84                 n = n + 1
 85             return dictData
 86 
 87     '''读取多条信息[根据条件]'''
 88     def fetchSome(self, tabel, condition, fields = '*', dataType = 0):
 89         field = isinstance(fields, list) and self.joinData(fields) or fields
 90         self.query('SELECT '+ field +' FROM ' + tabel + ' WHERE ' + self.parseMap(condition))
 91         tupleData = self.cur.fetchall()
 92         count = self.cur.rowcount
 93         if count > 0:
 94             if dataType == 0 or fields == '*' :
 95                 return (int(count), tupleData)
 96             else :
 97                 listData = []
 98                 for row in tupleData:
 99                     dictData = {}
100                     n = 0
101                     for k in fields:
102                         dictData[k] = row[n]
103                         n = n + 1
104                     listData.append(dictData)
105                 return (int(count), listData)
106         else:
107             return False
108 
109     '''获取信息总数[根据条件]'''
110     def getCount(self, tabel, condition = 0):
111         where = isinstance(condition, dict) and ' WHERE ' + self.parseMap(condition) or ''
112         self.query('SELECT 0 FROM ' + tabel + where)
113         return self.cur.rowcount
114 
115     '''提交事务'''
116     def commit(self):
117         self.conn.commit()
118 
119     '''关闭句柄和连接'''
120     def close(self):
121         self.cur.close()
122         self.conn.close()

DEMO

 1 # Filename: test_mysql.py
 2 
 3 from mysql_class import MyDb
 4 
 5 '''测试开始'''
 6 
 7 '''连接数据库[实例化]'''
 8 my_db = MyDb('127.0.0.1', 'python_user', '123123')
 9 
10 '''选择数据库'''
11 my_db.selectDb('test')
12 
13 
14 '''修改单条信息
15 dictData = {'class' : 'DD', 'name' : 'Pitt.C', 'subject' : 'Match_01', 'score' : 60}
16 condition = [{'class' : 'DD', 'name' : 'Tom'}, 'OR']
17 print my_db.saveOne('cla_info', dictData, condition)'''
18 
19 '''删除信息
20 condition = [{'class':'DD', 'name':'bd', 'id':17}, 'OR']
21 print my_db.deleteOne('cla_info', condition)'''
22 
23 '''获取单条信息
24 fields = ['class', 'name', 'subject', 'score', 'comment']
25 condition = {'id':6}
26 print my_db.fetchOne('cla_info', condition, fields)'''
27 
28 '''新增[单条]
29 dictData = {'class' : 'DDD', 'name' : 'Pitt', 'subject' : 'Match', 'score' : 97}
30 in_id = my_db.addOne('cla_info', dictData)
31 print 'Successful add data: ID(%d)' % in_id'''
32 
33 '''查询数据
34 field = ['class', 'name', 'subject', 'score', 'comment']
35 condition = {'name':'小明'}
36 ary = my_db.fetchSome('cla_info', condition, field, 1)
37 if ary or False :
38     print 'The total is:',ary[0]
39     print ary[1]'''
40 
41 '''获取总数
42 condition = {'name':'小明'}
43 print my_db.getCount('cla_info')'''
44 
45 '''事务处理'''
46 my_db.commit()
47 
48 '''关闭句柄和连接'''
49 my_db.close()