首页 > 代码库 > Python3 basedao.py的优化
Python3 basedao.py的优化
之前一篇Python 封装DBUtils 和pymysql 中写过一个basedao.py，最近几天又重新整理了下思绪，优化了下 basedao.py，目前支持的方法还不多，后续会进行改进、添加。
主要功能：
1.查询单个对象：
所需参数：表名，过滤条件
2.查询多个对象：
所需参数：表名，过滤条件
3.按主键查询：
所需参数：表名，值
4.分页查询：
所需参数：表名，页码，每页记录数，过滤条件
具体代码如下：
1 import json, os, sys, time 2 3 import pymysql 4 from DBUtils import PooledDB 5 6 class BaseDao(object): 7 """ 8 简便的数据库操作基类 9 """ 10 __config = {} # 数据库连接配置 11 __conn = None # 数据库连接 12 __cursor = None # 数据库游标 13 __database = None # 用于临时村塾查询数据库 14 __tableName = None # 用于临时存储查询表名 15 __fields = [] # 用于临时存储查询表的字段列表 16 __primaryKey_dict = {} # 用于存储配置中的数据库中所有表的主键 17 18 def __init__(self, creator=pymysql, host="localhost", user=None, password="", database=None, port=3306, charset="utf8"): 19 if host is None: 20 raise Exception("Parameter [host] is None.") 21 if user is None: 22 raise Exception("Parameter [user] is None.") 23 if password is None: 24 raise Exception("Parameter [password] is None.") 25 if database is None: 26 raise Exception("Parameter [database] is None.") 27 if port is None: 28 raise Exception("Parameter [port] is None.") 29 self.__config = dict({ 30 "creator" : creator, "charset":charset, 31 "host":host, "port":port, 32 "user":user, "password":password, "database":database 33 }) 34 self.__conn = PooledDB.connect(**self.__config) 35 self.__cursor = self.__conn.cursor() 36 self.__database = self.__config["database"] 37 self.__init_primaryKey() 38 print(get_time(), "数据库连接初始化成功。") 39 40 def __del__(self): 41 ‘重写类被清除时调用的方法‘ 42 if self.__cursor: 43 self.__cursor.close() 44 print(get_time(), "游标关闭") 45 if self.__conn: 46 self.__conn.close() 47 print(get_time(), "连接关闭") 48 49 def select_one(self, tableName=None, filters={}): 50 ‘‘‘ 51 查询单个对象 52 @tableName 表名 53 @filters 过滤条件 54 @return 返回字典集合，集合中以表字段作为 key，字段值作为 value 55 ‘‘‘ 56 self.__check_params(tableName) 57 sql = self.__query_util(filters) 58 self.__cursor.execute(sql) 59 result = self.__cursor.fetchone() 60 return self.__parse_result(result) 61 62 def select_pk(self, tableName=None, primaryKey=None): 63 ‘‘‘ 64 按主键查询 65 @tableName 表名 66 @primaryKey 主键值 67 ‘‘‘ 68 self.__check_params(tableName) 69 filters = {} 70 filters.setdefault(str(self.__primaryKey_dict[tableName]), primaryKey) 71 sql = self.__query_util(filters) 72 self.__cursor.execute(sql) 73 result = self.__cursor.fetchone() 74 return self.__parse_result(result) 75 76 def select_all(self, tableName=None, filters={}): 77 ‘‘‘ 78 查询所有 79 @tableName 表名 80 @filters 过滤条件 81 @return 返回字典集合，集合中以表字段作为 key，字段值作为 value 82 ‘‘‘ 83 self.__check_params(tableName) 84 sql = self.__query_util(filters) 85 self.__cursor.execute(sql) 86 results = self.__cursor.fetchall() 87 return self.__parse_results(results) 88 89 def count(self, tableName=None): 90 ‘‘‘ 91 统计记录数 92 ‘‘‘ 93 self.__check_params(tableName) 94 sql = "SELECT count(*) FROM %s"%(self.__tableName) 95 self.__cursor.execute(sql) 96 result = self.__cursor.fetchone() 97 return result[0] 98 99 def select_page(self, tableName=None, pageNum=1, limit=10, filters={}): 100 ‘‘‘ 101 分页查询 102 @tableName 表名 103 @return 返回字典集合，集合中以表字段作为 key，字段值作为 value 104 ‘‘‘ 105 self.__check_params(tableName) 106 totalCount = self.count() 107 if totalCount / limit == 0 : 108 totalPage = totalCount / limit 109 else: 110 totalPage = totalCount // limit + 1 111 if pageNum > totalPage: 112 print("最大页数为%d"%totalPage) 113 pageNum = totalPage 114 elif pageNum < 1: 115 print("页数不能小于1") 116 pageNum = 1 117 beginindex = (pageNum-1) * limit 118 filters.setdefault("_limit_", (beginindex, limit)) 119 sql = self.__query_util(filters) 120 self.__cursor.execute(sql) 121 results = self.__cursor.fetchall() 122 return self.__parse_results(results) 123 124 def __parse_result(self, result): 125 ‘用于解析单个查询结果，返回字典对象‘ 126 obj = {} 127 for k,v in zip(self.__fields, result): 128 obj[k] = v 129 return obj 130 131 def __parse_results(self, results): 132 ‘用于解析多个查询结果，返回字典列表对象‘ 133 objs = [] 134 for result in results: 135 obj = self.__parse_result(result) 136 objs.append(obj) 137 return objs 138 139 def __init_primaryKey(self): 140 ‘根据配置中的数据库读取该数据库中所有表的主键集合‘ 141 sql = """SELECT TABLE_NAME, COLUMN_NAME 142 FROM Information_schema.columns 143 WHERE COLUMN_KEY=‘PRI‘ AND TABLE_SCHEMA=‘%s‘"""%(self.__database) 144 self.__cursor.execute(sql) 145 results = self.__cursor.fetchall() 146 for result in results: 147 self.__primaryKey_dict[result[0]] = result[1] 148 149 def __query_fields(self, tableName=None, database=None): 150 ‘查询表的字段列表, 将查询出来的字段列表存入 __fields 中‘ 151 sql = """SELECT column_name 152 FROM Information_schema.columns 153 WHERE table_Name = ‘%s‘ AND TABLE_SCHEMA=‘%s‘"""%(tableName, database) 154 self.__cursor.execute(sql) 155 fields_tuple = self.__cursor.fetchall() 156 self.__fields = [fields[0] for fields in fields_tuple] 157 158 def __query_util(self, filters=None): 159 """ 160 SQL 语句拼接方法 161 @filters 过滤条件 162 """ 163 sql = r‘SELECT #{FIELDS} FROM #{TABLE_NAME} WHERE 1=1 #{FILTERS}‘ 164 # 拼接查询表 165 sql = sql.replace("#{TABLE_NAME}", self.__tableName) 166 # 拼接查询字段 167 self.__query_fields(self.__tableName, self.__database) 168 FIELDS = "" 169 for field in self.__fields: 170 FIELDS += field + ", " 171 FIELDS = FIELDS[0: len(FIELDS)-2] 172 sql = sql.replace("#{FIELDS}", FIELDS) 173 # 拼接查询条件（待优化） 174 if filters is None: 175 sql = sql.replace("#{FILTERS}", "") 176 else: 177 FILTERS = "" 178 if not isinstance(filters, dict): 179 raise Exception("Parameter [filters] must be dict type. ") 180 isPage = False 181 if filters.get("_limit_"): 182 isPage = True 183 beginindex, limit = filters.get("_limit_") 184 for k, v in filters.items(): 185 if k.startswith("_in_"): # 拼接 in 186 FILTERS += "AND %s IN (" %(k[4:]) 187 values = v.split(",") 188 for value in values: 189 FILTERS += "%s,"%value 190 FILTERS = FILTERS[0:len(FILTERS)-1] + ") " 191 elif k.startswith("_nein_"): # 拼接 not in 192 FILTERS += "AND %s NOT IN (" %(k[4:]) 193 values = v.split(",") 194 for value in values: 195 FILTERS += "%s,"%value 196 FILTERS = FILTERS[0:len(FILTERS)-1] + ") " 197 elif k.startswith("_like_"): # 拼接 like 198 FILTERS += "AND %s like ‘%%%s%%‘ " %(k[6:], v) 199 elif k.startswith("_ne_"): # 拼接不等于 200 FILTERS += "AND %s != ‘%s‘ " %(k[4:], v) 201 elif k.startswith("_lt_"): # 拼接小于 202 FILTERS += "AND %s < ‘%s‘ " %(k[4:], v) 203 elif k.startswith("_le_"): # 拼接小于等于 204 FILTERS += "AND %s <= ‘%s‘ " %(k[4:], v) 205 elif k.startswith("_gt_"): # 拼接大于 206 FILTERS += "AND %s > ‘%s‘ " %(k[4:], v) 207 elif k.startswith("_ge_"): # 拼接大于等于 208 FILTERS += "AND %s >= ‘%s‘ " %(k[4:], v) 209 elif k in self.__fields: # 拼接等于 210 FILTERS += "AND %s = ‘%s‘ "%(k, v) 211 sql = sql.replace("#{FILTERS}", FILTERS) 212 if isPage: 213 sql += "LIMIT %d,%d"%(beginindex, limit) 214 215 print(get_time(), sql) 216 return sql 217 218 def __check_params(self, tableName): 219 ‘‘‘ 220 检查参数 221 ‘‘‘ 222 if tableName: 223 self.__tableName = tableName 224 else: 225 if self.__tableName is None: 226 raise Exception("Parameter [tableName] is None.") 227 228 def get_time(): 229 return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) 230 231 if __name__ == "__main__": 232 config = { 233 # "creator": pymysql, 234 # "host" : "127.0.0.1", 235 "user" : "root", 236 "password" : "root", 237 "database" : "test", 238 # "port" : 3306, 239 # "charset" : ‘utf8‘ 240 } 241 base = BaseDao(**config) 242 ######################################################################## 243 # user = base.select_one("user") 244 # print(user) 245 ######################################################################## 246 # users = base.select_all("user") 247 # print(users) 248 ######################################################################## 249 # filter1 = { 250 # "sex":0, 251 # "_in_id":"1,2,3,4,5", 252 # "_like_name":"zhang", 253 # "_ne_name":"wangwu" 254 # } 255 # user_filters = base.select_all(tableName="user", filters=filter1) 256 # print(user_filters) 257 ######################################################################## 258 # menu = base.select_one(tableName="menu") 259 # print(menu) 260 ######################################################################## 261 # user_pk = base.select_pk("user", 2) 262 # print(user_pk) 263 ######################################################################## 264 # filter2 = { 265 # "_in_id":"1,2,3,4", 266 # "_like_name":"test" 267 # } 268 # user_limit = base.select_page("user", 2, 10, filter2) #未实现 269 # print(user_limit) 270 ########################################################################
代码中已经给出了几个具体示例，大家可以参考使用。
如果有感兴趣一起学习、讨论Python的可以加QQ群：626787819，有啥意见或者建议的可以发我邮箱：410093793@qq.com。
Python3 basedao.py的优化
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。