test_framework
/
AutomationTest
mirror of https://github.com/yanchunhuo/AutomationTest.git
4
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

数据库操作异常支持默认回滚

This commit is contained in:
yanchunhuo 2023-03-28 13:14:38 +08:00
parent c19e6212da
commit 7452095f98
1 changed files with 79 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

126
common_projects/sqliteOpt/base_db.py
View File

@ -3,7 +3,7 @@
# @author yanchunhuo # @author yanchunhuo
# @description # @description
# @created 2022-08-04T09:52:09.525Z+08:00 # @created 2022-08-04T09:52:09.525Z+08:00
# @last-modified 2022-09-05T19:07:04.273Z+08:00 # @last-modified 2023-03-28T13:14:23.295Z+08:00
# github https://github.com/yanchunhuo # github https://github.com/yanchunhuo
class Base_DB: class Base_DB:
@ -26,11 +26,15 @@ class Base_DB:
""" """
attrs=(obj.__dict__).copy() attrs=(obj.__dict__).copy()
attrs.pop('_sa_instance_state') attrs.pop('_sa_instance_state')
if order_by_columns: try:
result_object=self.db_session.query(self.model).filter_by(**attrs).order_by(*order_by_columns).first() if order_by_columns:
else: result_object=self.db_session.query(self.model).filter_by(**attrs).order_by(*order_by_columns).first()
result_object=self.db_session.query(self.model).filter_by(**attrs).first() else:
self.db_session.commit() result_object=self.db_session.query(self.model).filter_by(**attrs).first()
self.db_session.commit()
except:
self.db_session.rollback()
raise
return result_object return result_object
def filter_objects(self,obj:object,page_index:int=None,page_size:int=None,order_by_columns:list=None): def filter_objects(self,obj:object,page_index:int=None,page_size:int=None,order_by_columns:list=None):
@ -50,39 +54,59 @@ class Base_DB:
""" """
attrs=(obj.__dict__).copy() attrs=(obj.__dict__).copy()
attrs.pop('_sa_instance_state') attrs.pop('_sa_instance_state')
if page_index and page_size: try:
if order_by_columns: if page_index and page_size:
result_objects=self.db_session.query(self.model).filter_by(**attrs).\ if order_by_columns:
order_by(*order_by_columns).limit(page_size).offset((page_index-1)*page_size).all() result_objects=self.db_session.query(self.model).filter_by(**attrs).\
order_by(*order_by_columns).limit(page_size).offset((page_index-1)*page_size).all()
else:
result_objects=self.db_session.query(self.model).filter_by(**attrs).\
limit(page_size).offset((page_index-1)*page_size).all()
else: else:
result_objects=self.db_session.query(self.model).filter_by(**attrs).\ if order_by_columns:
limit(page_size).offset((page_index-1)*page_size).all() result_objects=self.db_session.query(self.model).filter_by(**attrs).order_by(*order_by_columns).all()
else: else:
if order_by_columns: result_objects=self.db_session.query(self.model).filter_by(**attrs).all()
result_objects=self.db_session.query(self.model).filter_by(**attrs).order_by(*order_by_columns).all() self.db_session.commit()
else: except:
result_objects=self.db_session.query(self.model).filter_by(**attrs).all() self.db_session.rollback()
self.db_session.commit() raise
return result_objects return result_objects
def add_object(self,obj:object): def add_object(self,obj:object):
self.db_session.add(obj) try:
self.db_session.commit() self.db_session.add(obj)
self.db_session.commit()
except:
self.db_session.rollback()
raise
def add_objects(self,objs:list=[]): def add_objects(self, objs: list = []):
self.db_session.add_all(objs) try:
self.commit() self.db_session.bulk_save_objects(objs)
self.db_session.commit()
except:
self.db_session.rollback()
raise
def delete_object(self,obj:object): def delete_object(self,obj:object):
attrs=(obj.__dict__).copy() attrs=(obj.__dict__).copy()
attrs.pop('_sa_instance_state') attrs.pop('_sa_instance_state')
num=self.db_session.query(self.model).filter_by(**attrs).delete() try:
self.db_session.commit() num=self.db_session.query(self.model).filter_by(**attrs).delete()
self.db_session.commit()
except:
self.db_session.rollback()
raise
return num return num
def empty_table(self): def empty_table(self):
num=self.db_session.query(self.model).delete() try:
self.db_session.commit() num=self.db_session.query(self.model).delete()
self.db_session.commit()
except:
self.db_session.rollback()
raise
return num return num
def update_object(self,old_obj:object,new_obj): def update_object(self,old_obj:object,new_obj):
@ -94,16 +118,20 @@ class Base_DB:
""" """
old_attrs=(old_obj.__dict__).copy() old_attrs=(old_obj.__dict__).copy()
old_attrs.pop('_sa_instance_state') old_attrs.pop('_sa_instance_state')
old_result_object=self.db_session.query(self.model).filter_by(**old_attrs).first() try:
if old_result_object: old_result_object=self.db_session.query(self.model).filter_by(**old_attrs).first()
for key in old_result_object.__dict__.keys(): if old_result_object:
if not key=='_sa_instance_state' and hasattr(new_obj,key): for key in old_result_object.__dict__.keys():
key_new_value=getattr(new_obj,key) if not key=='_sa_instance_state' and hasattr(new_obj,key):
if not key_new_value is None: key_new_value=getattr(new_obj,key)
if key_new_value=='null'.lower(): if not key_new_value is None:
key_new_value=None if key_new_value=='null'.lower():
setattr(old_result_object,key,key_new_value) key_new_value=None
self.db_session.commit() setattr(old_result_object,key,key_new_value)
self.db_session.commit()
except:
self.db_session.rollback()
raise
def update_objects(self,old_obj:object,new_obj): def update_objects(self,old_obj:object,new_obj):
"""变更的new_obj，如果需要将某个字段更新为Null，则设置字段值为'null' """变更的new_obj，如果需要将某个字段更新为Null，则设置字段值为'null'
@ -114,13 +142,17 @@ class Base_DB:
""" """
old_attrs=(old_obj.__dict__).copy() old_attrs=(old_obj.__dict__).copy()
old_attrs.pop('_sa_instance_state') old_attrs.pop('_sa_instance_state')
old_result_objects=self.db_session.query(self.model).filter_by(**old_attrs).all() try:
for old_result_object in old_result_objects: old_result_objects=self.db_session.query(self.model).filter_by(**old_attrs).all()
for key in old_result_object.__dict__.keys(): for old_result_object in old_result_objects:
if not key=='_sa_instance_state' and hasattr(new_obj,key): for key in old_result_object.__dict__.keys():
key_new_value=getattr(new_obj,key) if not key=='_sa_instance_state' and hasattr(new_obj,key):
if not key_new_value is None: key_new_value=getattr(new_obj,key)
if key_new_value=='null'.lower(): if not key_new_value is None:
key_new_value=None if key_new_value=='null'.lower():
setattr(old_result_object,key,key_new_value) key_new_value=None
self.db_session.commit() setattr(old_result_object,key,key_new_value)
self.db_session.commit()
except:
self.db_session.rollback()
raise