我的第一个python web开发框架(31)——定制ORM(七)

时间:2023-03-09 00:32:15
我的第一个python web开发框架(31)——定制ORM(七)

  几个复杂的ORM方式都已介绍完了,剩下一些常用的删除、获取记录数量、统计合计数、获取最大值、获取最小值等方法我就不一一详细介绍了,直接给出代码大家自行查看。

 #!/usr/bin/env python
# coding=utf-8 from common import db_helper class LogicBase():
"""逻辑层基础类""" def __init__(self, db, is_output_sql, table_name, column_name_list='*', pk_name='id'):
"""类初始化"""
# 数据库参数
self.__db = db
# 是否输出执行的Sql语句到日志中
self.__is_output_sql = is_output_sql
# 表名称
self.__table_name = str(table_name).lower()
# 查询的列字段名称,*表示查询全部字段,多于1个字段时用逗号进行分隔,除了字段名外,也可以是表达式
self.__column_name_list = str(column_name_list).lower()
# 主健名称
self.__pk_name = str(pk_name).lower() #####################################################################
### 执行Sql ### def select(self, sql):
"""执行sql查询语句(select)"""
with db_helper.PgHelper(self.__db, self.__is_output_sql) as db:
# 执行sql语句
result = db.execute(sql)
if not result:
result = []
return result def execute(self, sql):
"""执行sql语句,并提交事务"""
with db_helper.PgHelper(self.__db, self.__is_output_sql) as db:
# 执行sql语句
result = db.execute(sql)
if result:
db.commit()
else:
result = []
return result def copy(self, values, columns):
"""批量更新数据"""
with db_helper.PgHelper(self.__db, self.__is_output_sql) as db:
# 执行sql语句
result = db.copy(values, self.__table_name, columns)
return result def get_model(self, wheres):
"""通过条件获取一条记录"""
# 如果有条件,则自动添加where
if wheres:
wheres = ' where ' + wheres # 合成sql语句
sql = "select %(column_name_list)s from %(table_name)s %(wheres)s" % \
{'column_name_list': self.__column_name_list, 'table_name': self.__table_name, 'wheres': wheres}
# 初化化数据库链接
result = self.select(sql)
if result:
return result[0]
return {} def get_model_for_pk(self, pk, wheres=''):
"""通过主键值获取数据库记录实体"""
if not pk:
return {}
# 组装查询条件
wheres = '%s = %s' % (self.__pk_name, str(pk)) return self.get_model(wheres) def get_value(self, column_name, wheres=''):
"""
获取指定条件的字段值————多于条记录时,只取第一条记录
:param column_name: 单个字段名,如:id
:param wheres: 查询条件
:return: 7 (指定的字段值)
"""
if not column_name:
return None
elif wheres:
wheres = ' where ' + wheres sql = 'select %(column_name)s from %(table_name)s %(wheres)s limit 1' % \
{'column_name': column_name, 'table_name': self.__table_name, 'wheres': wheres}
result = self.select(sql)
# 如果查询成功,则直接返回记录字典
if result:
return result[0].get(column_name) def get_value_list(self, column_name, wheres=''):
"""
获取指定条件记录的字段值列表
:param column_name: 单个字段名,如:id
:param wheres: 查询条件
:return: [1,3,4,6,7]
"""
if not column_name:
column_name = self.__pk_name
elif wheres:
wheres = ' where ' + wheres sql = 'select array_agg(%(column_name)s) as list from %(table_name)s %(wheres)s' % \
{'column_name': column_name, 'table_name': self.__table_name, 'wheres': wheres}
result = self.select(sql)
# 如果查询失败或不存在指定条件记录,则直接返回初始值
if result and isinstance(result, list):
return result[0].get('list')
else:
return [] def add_model(self, fields, returning=''):
"""新增数据库记录"""
### 拼接sql语句 ###
# 初始化变量
key_list = []
value_list = []
# 将传入的字典参数进行处理,把字段名生成sql插入字段名数组和字典替换数组
# PS:字符串使用字典替换参数时,格式是%(name)s,这里会生成对应的字串
# 比如:
# 传入的字典为: {'id': 1, 'name': '名称'}
# 那么生成的key_list为:'id','name'
# 而value_list为:'%(id)s,%(name)s'
# 最终而value_list为字符串对应名称位置会被替换成相应的值
for key in fields.keys():
key_list.append(key)
value_list.append('%(' + key + ')s')
# 设置sql拼接字典,并将数组(lit)使用join方式进行拼接,生成用逗号分隔的字符串
parameter = {
'table_name': self.__table_name,
'pk_name': self.__pk_name,
'key_list': ','.join(key_list),
'value_list': ','.join(value_list)
}
# 如果有指定返回参数,则添加
if returning:
parameter['returning'] = ', ' + returning
else:
parameter['returning'] = '' # 生成可以使用字典替换的字符串
sql = "insert into %(table_name)s (%(key_list)s) values (%(value_list)s) returning %(pk_name)s %(returning)s" % parameter
# 将生成好的字符串替字典参数值,生成最终可执行的sql语句
sql = sql % fields result = self.execute(sql)
if result:
return result[0]
return {} def edit(self, fields, wheres='', returning=''):
"""批量编辑数据库记录"""
### 拼接sql语句 ###
# 拼接字段与值
field_list = [key + ' = %(' + key + ')s' for key in fields.keys()]
# 设置sql拼接字典
parameter = {
'table_name': self.__table_name,
'pk_name': self.__pk_name,
'field_list': ','.join(field_list)
}
# 如果存在更新条件,则将条件添加到sql拼接更换字典中
if wheres:
parameter['wheres'] = ' where ' + wheres
else:
parameter['wheres'] = '' # 如果有指定返回参数,则添加
if returning:
parameter['returning'] = ', ' + returning
else:
parameter['returning'] = '' # 生成sql语句
sql = "update %(table_name)s set %(field_list)s %(wheres)s returning %(pk_name)s %(returning)s" % parameter
sql = sql % fields return self.execute(sql) def edit_model(self, pk, fields, wheres='', returning=''):
"""编辑单条数据库记录"""
if not pk:
return {}
elif wheres:
wheres = self.__pk_name + ' = ' + str(pk) + ' and ' + wheres
else:
wheres = self.__pk_name + ' = ' + str(pk) return self.edit(fields, wheres, returning) def delete(self, wheres='', returning='', is_update_cache=True):
"""批量删除数据库记录"""
# 如果存在条件
if wheres:
wheres = ' where ' + wheres # 如果有指定返回参数,则添加
if returning:
returning = ', ' + returning # 生成sql语句
sql = "delete from %(table_name)s %(wheres)s returning %(pk_name)s %(returning)s" % \
{'table_name': self.__table_name, 'wheres': wheres, 'pk_name': self.__pk_name, 'returning': returning}
return self.execute(sql) def delete_model(self, pk, wheres='', returning='', is_update_cache=True):
"""删除单条数据库记录"""
if not pk:
return {}
elif wheres:
wheres = self.__pk_name + ' = ' + str(pk) + ' and ' + wheres
else:
wheres = self.__pk_name + ' = ' + str(pk) return self.delete(wheres, returning) def get_list(self, column_name_list='', wheres='', page_number=None, page_size=None, orderby=None, table_name=None):
"""
获取指定条件的数据库记录集
:param column_name_list: 查询字段
:param wheres: 查询条件
:param page_number: 分页索引值
:param page_size: 分页大小, 存在值时才会执行分页
:param orderby: 排序规则
:param table_name: 查询数据表,多表查询时需要设置
:return: 返回记录集总数量与分页记录集
{'records': 0, 'total': 0, 'page': 0, 'rows': []}
"""
# 初始化输出参数:总记录数量与列表集
data = {
'records': 0, # 总记录数
'total': 0, # 总页数
'page': 1, # 当前页面索引
'rows': [], # 查询结果(记录列表)
}
# 初始化查询数据表名称
if not table_name:
table_name = self.__table_name
# 初始化查询字段名
if not column_name_list:
column_name_list = self.__column_name_list
# 初始化查询条件
if wheres:
# 如果是字符串,表示该查询条件已组装好了,直接可以使用
if isinstance(wheres, str):
wheres = 'where ' + wheres
# 如果是list,则表示查询条件有多个,可以使用join将它们用and方式组合起来使用
elif isinstance(wheres, list):
wheres = 'where ' + ' and '.join(wheres)
# 初始化排序
if not orderby:
orderby = self.__pk_name + ' desc'
# 初始化分页查询的记录区间
paging = '' with db_helper.PgHelper(self.__db, self.__is_output_sql) as db:
#############################################################
# 判断是否需要进行分页
if not page_size is None:
### 执行sql,获取指定条件的记录总数量
sql = 'select count(1) as records from %(table_name)s %(wheres)s ' % \
{'table_name': table_name, 'wheres': wheres}
result = db.execute(sql)
# 如果查询失败或不存在指定条件记录,则直接返回初始值
if not result or result[0]['records'] == 0:
return data # 设置记录总数量
data['records'] = result[0].get('records') #########################################################
### 设置分页索引与页面大小 ###
if page_size <= 0:
page_size = 10
# 计算总分页数量:通过总记录数除于每页显示数量来计算总分页数量
if data['records'] % page_size == 0:
page_total = data['records'] // page_size
else:
page_total = data['records'] // page_size + 1
# 判断页码是否超出限制,超出限制查询时会出现异常,所以将页面索引设置为最后一页
if page_number < 1 or page_number > page_total:
page_number = page_total
# 记录总页面数量
data['total'] = page_total
# 记录当前页面值
data['page'] = page_number
# 计算当前页面要显示的记录起始位置(limit指定的位置)
record_number = (page_number - 1) * page_size
# 设置查询分页条件
paging = ' limit ' + str(page_size) + ' offset ' + str(record_number)
############################################################# ### 按条件查询数据库记录
sql = "select %(column_name_list)s from %(table_name)s %(wheres)s order by %(orderby)s %(paging)s" % \
{'column_name_list': column_name_list,
'table_name': table_name,
'wheres': wheres,
'orderby': orderby,
'paging': paging}
result = db.execute(sql)
if result:
data['rows'] = result
# 不需要分页查询时,直接在这里设置总记录数
if page_size is None:
data['records'] = len(result) return data def get_count(self, wheres=''):
"""获取指定条件记录数量"""
if wheres:
wheres = ' where ' + wheres
sql = 'select count(1) as total from %(table_name)s %(wheres)s ' % \
{'table_name': self.__table_name, 'wheres': wheres}
result = self.select(sql)
# 如果查询存在记录,则返回true
if result:
return result[0].get('total')
return 0 def get_sum(self, fields, wheres):
"""获取指定条件记录数量"""
sql = 'select sum(%(fields)s) as total from %(table_name)s where %(wheres)s ' % \
{'table_name': self.__table_name, 'wheres': wheres, 'fields': fields}
result = self.select(sql)
# 如果查询存在记录,则返回true
if result and result[0].get('total'):
return result[0].get('total')
return 0 def get_min(self, fields, wheres):
"""获取该列记录最小值"""
sql = 'select min(%(fields)s) as min from %(table_name)s where %(wheres)s ' % \
{'table_name': self.__table_name, 'wheres': wheres, 'fields': fields}
result = self.select(sql)
# 如果查询存在记录,则返回true
if result and result[0].get('min'):
return result[0].get('min') def get_max(self, fields, wheres):
"""获取该列记录最大值"""
sql = 'select max(%(fields)s) as max from %(table_name)s where %(wheres)s ' % \
{'table_name': self.__table_name, 'wheres': wheres, 'fields': fields}
result = self.select(sql)
# 如果查询存在记录,则返回true
if result and result[0].get('max'):
return result[0].get('max') #####################################################################

  大家只要掌握了ORM简单的组合sql方法,就可以*发挥,根据自己的需要去创建不同的方法了,也可以随意更换mysql、mssql等数据库。

  当然,这只是最简单的ORM方式,提交字段参数和条件参数时,它不会自动分辨字段的类型,不会自动初始化默认值,如果想让它变的更加强大,还需要做更多的改造与处理,这样做的话它也会跟着变的更加复杂和难懂,性能也会跟着下降。不过当前功能对于多数项目来说,已经足够使用了。大家如果有需要可以自行研究进行扩展。

  在日常操作中,获取指定记录实体是最常见使用最频繁的操作,为了减少对数据库的查询,我们可以将ORM与Nosql结合起来,提升ORM的操作性能,当然如果你不想使用nosql缓存,也可以直接跳过本章节。

  使用Nosql,首先我们需要一个链接Nosql的配置文件,用它来存储Nosql的服务地址、端口、密码等参数

  在config文件夹中我们创建redis_config.py配置文件

#!/usr/bin/env python
# coding=utf-8 ### redis缓存配置参数 ###
REDIS = {
# 服务地址
'server': '127.0.0.1',
# 服务端口
'post': 6379,
# 服务密码
'pwd': '',
# 数据库序号
'db': 1
}

  然后我们还需要一个nosql链接工具包(cache_helper.py),用来对nosql进行set、get、delete和clear操作(存储、获取、删除和清空缓存)

 #!/usr/bin/env python
# coding=utf-8 import redis from common import log_helper
from config import redis_config # 设置redis配置参数
_redis = redis_config.REDIS
# 初始化Redis缓存链接
r = None
try:
if not r:
r = redis.Redis(host=_redis.get('server', ''),
port=_redis.get('post', ''),
db=_redis.get('db', ''),
password=_redis.get('pwd', ''),
socket_timeout=1,
socket_connect_timeout=1)
except Exception as e:
log_helper.info('连接redis出错:(' + str(_redis) + ')' + str(e.args))
pass def set(key, value, time=86400):
"""
写缓存
:param key: 缓存key,字符串,不区分大小写
:param value: 要存储的值
:param time: 缓存过期时间(单位:秒),0=永不过期
:return:
"""
# 将key转换为小写字母
key = str(key).lower()
try:
r.set(key, value, time)
except Exception as e:
log_helper.info('写缓存失败:key(' + key + ')' + str(e.args))
pass def get(key):
"""
读缓存
:param key: 缓存key,字符串,不区分大小写
:return:
"""
# 将key转换为小写字母
key = str(key).lower()
try:
value = r.get(key)
except Exception as e:
# log_helper.error('读缓存失败:key(' + key + ')' + str(e.args) + ' r:' + str(r) + ' _redis:' + str(_redis))
value = None return _str_to_json(value) def push(key, value):
"""
添加数据到队列头部
:param key: 缓存key,字符串,不区分大小写
:param value: 要存储的值
"""
# 将key转换为小写字母
key = str(key).lower()
try:
r.lpush(key, value)
except Exception as e:
log_helper.info('写缓存失败:key(' + key + ')' + str(e.args))
pass def pop(key):
"""
从缓存队列的后尾读取一条数据
:param key: 缓存key,字符串,不区分大小写
:return: 缓存数据
"""
# 将key转换为小写字母
key = str(key).lower()
try:
value = r.rpop(key)
except Exception as e:
log_helper.info('读取缓存队列失败:key(' + key + ')' + str(e.args))
value = None return _str_to_json(value) def _str_to_json(value):
"""
将缓存中读取出来的字符串转换成对应的数据、元组、列表或字典
"""
if not value:
return value
# 否则直接转换
try:
value = value.decode()
return eval(value)
except Exception as e:
print(e.args)
pass
# 否则直接输出字符串
return value def delete(key):
"""
删除缓存
:param key:缓存key,字符串,不区分大小写
:return:
"""
# 将key转换为小写字母
key = str(key).lower()
try:
log_helper.info(str(r.delete(key)))
except Exception as e:
log_helper.info('Exception:' + str(e.args))
pass def clear():
"""
清空所有缓存
"""
try:
r.flushdb()
except:
pass

  我常用的是redis,所以使用cache_helper.py时,需要安装redis服务和对应的Python包。如果你使用的是memcache,你只需要重构一下cache_helper.py代码就可以了。

  接下来我们改造一下逻辑层基类(ORM模块)

  首先我们需要导入cache_helper

from common import db_helper, cache_helper

  在使用nosql缓存时,大家都知道我们是使用key来进行对象存取的,而这个key也是唯一的,所以key的生成就很重要的,为了避免key的重复,我们在对记录设置key时,可以用表名+主键id的方式来组合key,当然为了调用方便,可以将获取key写成一个方法来生成

    def get_cache_key(self, pk):
"""获取缓存key值"""
return ''.join((self.__table_name, '_', str(pk)))

  这里使用join的方法,将表名、下横线、主键值组合生成缓存key字符串

  对于缓存的操作,主要有设置缓存、获取缓存、删除缓存这三种操作,当然为了方便我们获取记录中指定字段值,我们可以增加读取指定字段值方法。

  首先是设置缓存方法,大家看看下面代码,它非常简单,先调用生成缓存key,然后将对象存储到缓存中,并指定过期时间,当设置time为0时,它将永不过期

    def set_model_for_cache(self, pk, value, time=43200):
"""更新存储在缓存中的数据库记录,缓存过期时间为12小时"""
# 生成缓存key
key = self.get_cache_key(pk)
# 存储到nosql缓存中
cache_helper.set(key, value, time)

  接着是获取缓存对象方法

    def get_model_for_cache(self, pk):
"""从缓存中读取数据库记录"""
# 生成缓存key
key = self.get_cache_key(pk)
# 从缓存中读取数据库记录
result = cache_helper.get(key)
# 缓存中不存在记录,则从数据库获取
if not result:
result = self.get_model_for_pk(pk)
self.set_model_for_cache(pk, result)
if result:
return result
else:
return {}

  我们首先要做的同样是生成缓存key,然后调用get方法从缓存中读取对象,执行完后,需要判断该对象是否存在缓存中,如果不存在则表示该对象并未存储到缓存中或它可能存储过期了,所以需要重新从数据库中读取出来,并将它存储到缓存中,然后将读取出来的记录实体返回出去。

  然后我们再增加一个读取指定记录字段值的方法

    def get_value_for_cache(self, pk, column_name):
"""获取指定记录的字段值"""
return self.get_model_for_cache(pk).get(column_name)

  它直接调用获取缓存对象方法,然后从返回的对象中读取指定的字段值就可以了

  删除缓存方法也很简单,生成缓存key后,直接调用delete进行删除。对于删除方法,有时候调用不知是不是nosql自身bug问题,还是在主从关系的nosql中读写分离会引起删除失败,如果出现这种情况,可以将delete改为set,将该缓存set为空就可以解决这个问题

    def del_model_for_cache(self, pk):
"""删除缓存中指定数据"""
# 生成缓存key
key = self.get_cache_key(pk)
# log_helper.info(key)
# 存储到nosql缓存中
cache_helper.delete(key)

  PS:在使用缓存操作时,有时我们直接对数据库进行操作,就会引起数据与缓存不匹配,出现脏数据的情况,这时在后台增加清空缓存的操作,直接调用cache_helper.clear()进行清空缓存。

  基本方法都完成了,接下来就是要对ORM的删除与修改方法进行改造了,让它们自行根据需要对缓存进行对应操作,让缓存与数据表中的记录保持一致。

  在改造时,我们只需要对删除与修改操作进行处理,对新增与查询操作不需要操作,因为新增的记录,它并在缓存中并不存在,所以不需要进行操作,而查询也不会改变数据内容,只有进行删除和修改操作时,才会变动数据内容,这时就需要更改缓存,让数据保持一致。

  改造编辑记录实体方法

     def edit(self, fields, wheres='', returning='', is_update_cache=True):
"""
批量编辑数据库记录
:param fields: 要更新的字段(字段名与值存储在字典中)
:param wheres: 更新条件
:param returning: 更新成功后,返回的字段名
:param is_update_cache: 是否同步更新缓存
:return:
"""
### 拼接sql语句 ###
# 拼接字段与值
field_list = [key + ' = %(' + key + ')s' for key in fields.keys()]
# 设置sql拼接字典
parameter = {
'table_name': self.__table_name,
'pk_name': self.__pk_name,
'field_list': ','.join(field_list)
}
# 如果存在更新条件,则将条件添加到sql拼接更换字典中
if wheres:
parameter['wheres'] = ' where ' + wheres
else:
parameter['wheres'] = '' # 如果有指定返回参数,则添加
if returning:
parameter['returning'] = ', ' + returning
else:
parameter['returning'] = '' # 生成sql语句
sql = "update %(table_name)s set %(field_list)s %(wheres)s returning %(pk_name)s %(returning)s" % parameter
sql = sql % fields result = self.execute(sql)
if result:
# 判断是否删除对应的缓存
if is_update_cache:
# 循环删除更新成功的所有记录对应的缓存
for model in result:
self.del_model_for_cache(model.get('id', 0))
return result

  大家可以看到,该方法增加了is_update_cache 是否同步更新缓存参数,这是因为我们在使用缓存时会存在一些特殊情况,比如说批量更新很多数据时,如果使用循环逐条清理对应缓存时,会占用较多资源,我们可以关掉缓存的同步更新,直接调用clear清空所有缓存会更加快捷;又比如说,页面访问数的更新,它会更新的非常频繁,我们不需要实时清除,可以使用其他方式触发清理,也可以将点击数用独立缓存存储使用等

  而清理缓存,我们只需要将缓存内容直接删除就可以了,因为执行更新以后,返回的记录实体没有设置为*时,只返回主键id,直接设置的话会造成缓存数据丢失细节的问题,另外,我们执行更新以后,该记录也不一定还会被读取出来。

  删除记录也进行一样的改造

     def delete(self, wheres='', returning='', is_update_cache=True):
"""
批量删除数据库记录
:param wheres: 删除条件
:param returning: 删除成功后,返回的字段名
:param is_update_cache: 是否同步更新缓存
:return:
"""
# 如果存在条件
if wheres:
wheres = ' where ' + wheres # 如果有指定返回参数,则添加
if returning:
returning = ', ' + returning # 生成sql语句
sql = "delete from %(table_name)s %(wheres)s returning %(pk_name)s %(returning)s" % \
{'table_name': self.__table_name, 'wheres': wheres, 'pk_name': self.__pk_name, 'returning': returning}
result = self.execute(sql)
if result:
# 同步删除对应的缓存
if is_update_cache:
for model in result:
self.del_model_for_cache(model.get('id', 0))
return result

  对于缓存基本上就这两个要进行改造的操作了。在实现开发中,我们认真想一想,其实我们还会存在一些特殊的情况,比如说我们对数据进行加工处理后,将加工后的值存储到缓存中,而对相关记录进行修改或删除操作以后,由于这些缓存它与记录并没有关联,所以执行相关操作以后,它就变成孤岛,不会实时同步,产生脏数据。所以我们需要有一个功能,可以将它们管理起来,与该数据表的修改和删除操作关联起来,进行修改和删除操作后同步清除这些特殊缓存。

  根据这些要求,我们就需要再增加两个缓存操作方法,用来存储这些特殊的缓存名称,然后在进行修改和删除操作时,同步清除这些特殊缓存。

  首先我们需要在初始化方法中,添加一个绑定该数据表的全局缓存变量self.__cache_list,它由表名称+_cache_list组成。

     def __init__(self, db, is_output_sql, table_name, column_name_list='*', pk_name='id'):
"""类初始化"""
# 数据库参数
self.__db = db
# 是否输出执行的Sql语句到日志中
self.__is_output_sql = is_output_sql
# 表名称
self.__table_name = str(table_name).lower()
# 查询的列字段名称,*表示查询全部字段,多于1个字段时用逗号进行分隔,除了字段名外,也可以是表达式
self.__column_name_list = str(column_name_list).lower()
# 主健名称
self.__pk_name = str(pk_name).lower()
# 缓存列表
self.__cache_list = self.__table_name + '_cache_list'

  然后我们再添加特殊缓存存储方法

     def add_relevance_cache_in_list(self, key):
"""将缓存名称存储到列表里————主要存储与记录变更关联的"""
# 从nosql中读取全局缓存列表
cache_list = cache_helper.get(self.__cache_list)
# 判断缓存列表是否有值,有则进行添加操作
if cache_list:
# 判断是否已存储列表中,不存在则执行添加操作
if not key in cache_list:
cache_list.append(key)
cache_helper.set(self.__cache_list, cache_list)
# 无则直接创建全局缓存列表,并存储到nosql中
else:
cache_list = [key]
cache_helper.set(self.__cache_list, cache_list)

  执行该方法,会将我们自定义的缓存名称存储到全局缓存变量中

  接着我们再添加一个清除所有特殊缓存的方法

     def del_relevance_cache(self):
"""删除关联缓存————将和数据表记录关联的,个性化缓存全部删除"""
# 从nosql中读取全局缓存列表
cache_list = cache_helper.get(self.__cache_list)
# 清除已删除缓存列表
cache_helper.delete(self.__cache_list)
if cache_list:
# 执行删除操作
for cache in cache_list:
cache_helper.delete(cache)

  添加完成以后,我们再来改造一下修改与删除代码,只需要在里面添加清除所有特殊缓存方法就可以了

     def edit(self, fields, wheres='', returning='', is_update_cache=True):
"""
批量编辑数据库记录
:param fields: 要更新的字段(字段名与值存储在字典中)
:param wheres: 更新条件
:param returning: 更新成功后,返回的字段名
:param is_update_cache: 是否同步更新缓存
:return:
"""
### 拼接sql语句 ###
# 拼接字段与值
field_list = [key + ' = %(' + key + ')s' for key in fields.keys()]
# 设置sql拼接字典
parameter = {
'table_name': self.__table_name,
'pk_name': self.__pk_name,
'field_list': ','.join(field_list)
}
# 如果存在更新条件,则将条件添加到sql拼接更换字典中
if wheres:
parameter['wheres'] = ' where ' + wheres
else:
parameter['wheres'] = '' # 如果有指定返回参数,则添加
if returning:
parameter['returning'] = ', ' + returning
else:
parameter['returning'] = '' # 生成sql语句
sql = "update %(table_name)s set %(field_list)s %(wheres)s returning %(pk_name)s %(returning)s" % parameter
sql = sql % fields result = self.execute(sql)
if result:
# 判断是否删除对应的缓存
if is_update_cache:
# 循环删除更新成功的所有记录对应的缓存
for model in result:
self.del_model_for_cache(model.get('id', 0))
# 同步删除与本表关联的缓存
self.del_relevance_cache()
return result def delete(self, wheres='', returning='', is_update_cache=True):
"""
批量删除数据库记录
:param wheres: 删除条件
:param returning: 删除成功后,返回的字段名
:param is_update_cache: 是否同步更新缓存
:return:
"""
# 如果存在条件
if wheres:
wheres = ' where ' + wheres # 如果有指定返回参数,则添加
if returning:
returning = ', ' + returning # 生成sql语句
sql = "delete from %(table_name)s %(wheres)s returning %(pk_name)s %(returning)s" % \
{'table_name': self.__table_name, 'wheres': wheres, 'pk_name': self.__pk_name, 'returning': returning}
result = self.execute(sql)
if result:
# 同步删除对应的缓存
if is_update_cache:
for model in result:
self.del_model_for_cache(model.get('id', 0))
# 同步删除与本表关联的缓存
self.del_relevance_cache()
return result

  

  ORM的缓存改造就全部完成了,下面是完整代码

 #!/usr/bin/env python
# coding=utf-8 from common import db_helper, cache_helper class LogicBase():
"""逻辑层基础类""" def __init__(self, db, is_output_sql, table_name, column_name_list='*', pk_name='id'):
"""类初始化"""
# 数据库参数
self.__db = db
# 是否输出执行的Sql语句到日志中
self.__is_output_sql = is_output_sql
# 表名称
self.__table_name = str(table_name).lower()
# 查询的列字段名称,*表示查询全部字段,多于1个字段时用逗号进行分隔,除了字段名外,也可以是表达式
self.__column_name_list = str(column_name_list).lower()
# 主健名称
self.__pk_name = str(pk_name).lower()
# 缓存列表
self.__cache_list = self.__table_name + '_cache_list' #####################################################################
### 执行Sql ### def select(self, sql):
"""执行sql查询语句(select)"""
with db_helper.PgHelper(self.__db, self.__is_output_sql) as db:
# 执行sql语句
result = db.execute(sql)
if not result:
result = []
return result def execute(self, sql):
"""执行sql语句,并提交事务"""
with db_helper.PgHelper(self.__db, self.__is_output_sql) as db:
# 执行sql语句
result = db.execute(sql)
if result:
db.commit()
else:
result = []
return result def copy(self, values, columns):
"""批量更新数据"""
with db_helper.PgHelper(self.__db, self.__is_output_sql) as db:
# 执行sql语句
result = db.copy(values, self.__table_name, columns)
return result def get_model(self, wheres):
"""通过条件获取一条记录"""
# 如果有条件,则自动添加where
if wheres:
wheres = ' where ' + wheres # 合成sql语句
sql = "select %(column_name_list)s from %(table_name)s %(wheres)s" % \
{'column_name_list': self.__column_name_list, 'table_name': self.__table_name, 'wheres': wheres}
# 初化化数据库链接
result = self.select(sql)
if result:
return result[0]
return {} def get_model_for_pk(self, pk, wheres=''):
"""通过主键值获取数据库记录实体"""
if not pk:
return {}
# 组装查询条件
wheres = '%s = %s' % (self.__pk_name, str(pk)) return self.get_model(wheres) def get_value(self, column_name, wheres=''):
"""
获取指定条件的字段值————多于条记录时,只取第一条记录
:param column_name: 单个字段名,如:id
:param wheres: 查询条件
:return: 7 (指定的字段值)
"""
if not column_name:
return None
elif wheres:
wheres = ' where ' + wheres sql = 'select %(column_name)s from %(table_name)s %(wheres)s limit 1' % \
{'column_name': column_name, 'table_name': self.__table_name, 'wheres': wheres}
result = self.select(sql)
# 如果查询成功,则直接返回记录字典
if result:
return result[0].get(column_name) def get_value_list(self, column_name, wheres=''):
"""
获取指定条件记录的字段值列表
:param column_name: 单个字段名,如:id
:param wheres: 查询条件
:return: [1,3,4,6,7]
"""
if not column_name:
column_name = self.__pk_name
elif wheres:
wheres = ' where ' + wheres sql = 'select array_agg(%(column_name)s) as list from %(table_name)s %(wheres)s' % \
{'column_name': column_name, 'table_name': self.__table_name, 'wheres': wheres}
result = self.select(sql)
# 如果查询失败或不存在指定条件记录,则直接返回初始值
if result and isinstance(result, list):
return result[0].get('list')
else:
return [] def add_model(self, fields, returning=''):
"""新增数据库记录"""
### 拼接sql语句 ###
# 初始化变量
key_list = []
value_list = []
# 将传入的字典参数进行处理,把字段名生成sql插入字段名数组和字典替换数组
# PS:字符串使用字典替换参数时,格式是%(name)s,这里会生成对应的字串
# 比如:
# 传入的字典为: {'id': 1, 'name': '名称'}
# 那么生成的key_list为:'id','name'
# 而value_list为:'%(id)s,%(name)s'
# 最终而value_list为字符串对应名称位置会被替换成相应的值
for key in fields.keys():
key_list.append(key)
value_list.append('%(' + key + ')s')
# 设置sql拼接字典,并将数组(lit)使用join方式进行拼接,生成用逗号分隔的字符串
parameter = {
'table_name': self.__table_name,
'pk_name': self.__pk_name,
'key_list': ','.join(key_list),
'value_list': ','.join(value_list)
}
# 如果有指定返回参数,则添加
if returning:
parameter['returning'] = ', ' + returning
else:
parameter['returning'] = '' # 生成可以使用字典替换的字符串
sql = "insert into %(table_name)s (%(key_list)s) values (%(value_list)s) returning %(pk_name)s %(returning)s" % parameter
# 将生成好的字符串替字典参数值,生成最终可执行的sql语句
sql = sql % fields result = self.execute(sql)
if result:
return result[0]
return {} def edit(self, fields, wheres='', returning='', is_update_cache=True):
"""
批量编辑数据库记录
:param fields: 要更新的字段(字段名与值存储在字典中)
:param wheres: 更新条件
:param returning: 更新成功后,返回的字段名
:param is_update_cache: 是否同步更新缓存
:return:
"""
### 拼接sql语句 ###
# 拼接字段与值
field_list = [key + ' = %(' + key + ')s' for key in fields.keys()]
# 设置sql拼接字典
parameter = {
'table_name': self.__table_name,
'pk_name': self.__pk_name,
'field_list': ','.join(field_list)
}
# 如果存在更新条件,则将条件添加到sql拼接更换字典中
if wheres:
parameter['wheres'] = ' where ' + wheres
else:
parameter['wheres'] = '' # 如果有指定返回参数,则添加
if returning:
parameter['returning'] = ', ' + returning
else:
parameter['returning'] = '' # 生成sql语句
sql = "update %(table_name)s set %(field_list)s %(wheres)s returning %(pk_name)s %(returning)s" % parameter
sql = sql % fields result = self.execute(sql)
if result:
# 判断是否删除对应的缓存
if is_update_cache:
# 循环删除更新成功的所有记录对应的缓存
for model in result:
self.del_model_for_cache(model.get('id', 0))
# 同步删除与本表关联的缓存
self.del_relevance_cache()
return result def edit_model(self, pk, fields, wheres='', returning=''):
"""编辑单条数据库记录"""
if not pk:
return {}
elif wheres:
wheres = self.__pk_name + ' = ' + str(pk) + ' and ' + wheres
else:
wheres = self.__pk_name + ' = ' + str(pk) return self.edit(fields, wheres, returning) def delete(self, wheres='', returning='', is_update_cache=True):
"""
批量删除数据库记录
:param wheres: 删除条件
:param returning: 删除成功后,返回的字段名
:param is_update_cache: 是否同步更新缓存
:return:
"""
# 如果存在条件
if wheres:
wheres = ' where ' + wheres # 如果有指定返回参数,则添加
if returning:
returning = ', ' + returning # 生成sql语句
sql = "delete from %(table_name)s %(wheres)s returning %(pk_name)s %(returning)s" % \
{'table_name': self.__table_name, 'wheres': wheres, 'pk_name': self.__pk_name, 'returning': returning}
result = self.execute(sql)
if result:
# 同步删除对应的缓存
if is_update_cache:
for model in result:
self.del_model_for_cache(model.get('id', 0))
# 同步删除与本表关联的缓存
self.del_relevance_cache()
return result def delete_model(self, pk, wheres='', returning='', is_update_cache=True):
"""删除单条数据库记录"""
if not pk:
return {}
elif wheres:
wheres = self.__pk_name + ' = ' + str(pk) + ' and ' + wheres
else:
wheres = self.__pk_name + ' = ' + str(pk) return self.delete(wheres, returning) def get_list(self, column_name_list='', wheres='', page_number=None, page_size=None, orderby=None, table_name=None):
"""
获取指定条件的数据库记录集
:param column_name_list: 查询字段
:param wheres: 查询条件
:param page_number: 分页索引值
:param page_size: 分页大小, 存在值时才会执行分页
:param orderby: 排序规则
:param table_name: 查询数据表,多表查询时需要设置
:return: 返回记录集总数量与分页记录集
{'records': 0, 'total': 0, 'page': 0, 'rows': []}
"""
# 初始化输出参数:总记录数量与列表集
data = {
'records': 0, # 总记录数
'total': 0, # 总页数
'page': 1, # 当前页面索引
'rows': [], # 查询结果(记录列表)
}
# 初始化查询数据表名称
if not table_name:
table_name = self.__table_name
# 初始化查询字段名
if not column_name_list:
column_name_list = self.__column_name_list
# 初始化查询条件
if wheres:
# 如果是字符串,表示该查询条件已组装好了,直接可以使用
if isinstance(wheres, str):
wheres = 'where ' + wheres
# 如果是list,则表示查询条件有多个,可以使用join将它们用and方式组合起来使用
elif isinstance(wheres, list):
wheres = 'where ' + ' and '.join(wheres)
# 初始化排序
if not orderby:
orderby = self.__pk_name + ' desc'
# 初始化分页查询的记录区间
paging = '' with db_helper.PgHelper(self.__db, self.__is_output_sql) as db:
#############################################################
# 判断是否需要进行分页
if not page_size is None:
### 执行sql,获取指定条件的记录总数量
sql = 'select count(1) as records from %(table_name)s %(wheres)s ' % \
{'table_name': table_name, 'wheres': wheres}
result = db.execute(sql)
# 如果查询失败或不存在指定条件记录,则直接返回初始值
if not result or result[0]['records'] == 0:
return data # 设置记录总数量
data['records'] = result[0].get('records') #########################################################
### 设置分页索引与页面大小 ###
if page_size <= 0:
page_size = 10
# 计算总分页数量:通过总记录数除于每页显示数量来计算总分页数量
if data['records'] % page_size == 0:
page_total = data['records'] // page_size
else:
page_total = data['records'] // page_size + 1
# 判断页码是否超出限制,超出限制查询时会出现异常,所以将页面索引设置为最后一页
if page_number < 1 or page_number > page_total:
page_number = page_total
# 记录总页面数量
data['total'] = page_total
# 记录当前页面值
data['page'] = page_number
# 计算当前页面要显示的记录起始位置(limit指定的位置)
record_number = (page_number - 1) * page_size
# 设置查询分页条件
paging = ' limit ' + str(page_size) + ' offset ' + str(record_number)
############################################################# ### 按条件查询数据库记录
sql = "select %(column_name_list)s from %(table_name)s %(wheres)s order by %(orderby)s %(paging)s" % \
{'column_name_list': column_name_list,
'table_name': table_name,
'wheres': wheres,
'orderby': orderby,
'paging': paging}
result = db.execute(sql)
if result:
data['rows'] = result
# 不需要分页查询时,直接在这里设置总记录数
if page_size is None:
data['records'] = len(result) return data def get_count(self, wheres=''):
"""获取指定条件记录数量"""
if wheres:
wheres = ' where ' + wheres
sql = 'select count(1) as total from %(table_name)s %(wheres)s ' % \
{'table_name': self.__table_name, 'wheres': wheres}
result = self.select(sql)
# 如果查询存在记录,则返回true
if result:
return result[0].get('total')
return 0 def get_sum(self, fields, wheres):
"""获取指定条件记录数量"""
sql = 'select sum(%(fields)s) as total from %(table_name)s where %(wheres)s ' % \
{'table_name': self.__table_name, 'wheres': wheres, 'fields': fields}
result = self.select(sql)
# 如果查询存在记录,则返回true
if result and result[0].get('total'):
return result[0].get('total')
return 0 def get_min(self, fields, wheres):
"""获取该列记录最小值"""
sql = 'select min(%(fields)s) as min from %(table_name)s where %(wheres)s ' % \
{'table_name': self.__table_name, 'wheres': wheres, 'fields': fields}
result = self.select(sql)
# 如果查询存在记录,则返回true
if result and result[0].get('min'):
return result[0].get('min') def get_max(self, fields, wheres):
"""获取该列记录最大值"""
sql = 'select max(%(fields)s) as max from %(table_name)s where %(wheres)s ' % \
{'table_name': self.__table_name, 'wheres': wheres, 'fields': fields}
result = self.select(sql)
# 如果查询存在记录,则返回true
if result and result[0].get('max'):
return result[0].get('max') ##################################################################### #####################################################################
### 缓存操作方法 ### def get_cache_key(self, pk):
"""获取缓存key值"""
return ''.join((self.__table_name, '_', str(pk))) def set_model_for_cache(self, pk, value, time=43200):
"""更新存储在缓存中的数据库记录,缓存过期时间为12小时"""
# 生成缓存key
key = self.get_cache_key(pk)
# 存储到nosql缓存中
cache_helper.set(key, value, time) def get_model_for_cache(self, pk):
"""从缓存中读取数据库记录"""
# 生成缓存key
key = self.get_cache_key(pk)
# 从缓存中读取数据库记录
result = cache_helper.get(key)
# 缓存中不存在记录,则从数据库获取
if not result:
result = self.get_model_for_pk(pk)
self.set_model_for_cache(pk, result)
if result:
return result
else:
return {} def get_value_for_cache(self, pk, column_name):
"""获取指定记录的字段值"""
return self.get_model_for_cache(pk).get(column_name) def del_model_for_cache(self, pk):
"""删除缓存中指定数据"""
# 生成缓存key
key = self.get_cache_key(pk)
# log_helper.info(key)
# 存储到nosql缓存中
cache_helper.delete(key) def add_relevance_cache_in_list(self, key):
"""将缓存名称存储到列表里————主要存储与记录变更关联的"""
# 从nosql中读取全局缓存列表
cache_list = cache_helper.get(self.__cache_list)
# 判断缓存列表是否有值,有则进行添加操作
if cache_list:
# 判断是否已存储列表中,不存在则执行添加操作
if not key in cache_list:
cache_list.append(key)
cache_helper.set(self.__cache_list, cache_list)
# 无则直接创建全局缓存列表,并存储到nosql中
else:
cache_list = [key]
cache_helper.set(self.__cache_list, cache_list) def del_relevance_cache(self):
"""删除关联缓存————将和数据表记录关联的,个性化缓存全部删除"""
# 从nosql中读取全局缓存列表
cache_list = cache_helper.get(self.__cache_list)
# 清除已删除缓存列表
cache_helper.delete(self.__cache_list)
if cache_list:
# 执行删除操作
for cache in cache_list:
cache_helper.delete(cache) #####################################################################

版权声明:本文原创发表于 博客园,作者为 AllEmpty 本文欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接,否则视为侵权。

python开发QQ群:669058475(本群已满)、733466321(可以加2群)    作者博客:http://www.cnblogs.com/EmptyFS/