Flask-SQLAlchemy 操作数据库十分的方便，但也不影响我们再为它封装一个数据库操作基类，
patch 包是用于存放被覆写的模块，在其中新建 db.py，分别覆写 SQLAlchemy、Pagination、BaseQuery，使得其支持数据库操作自动提交，查询未被软删的记录，以及在业务上隐藏和添加字段，特别像 Django 序列号器的隐藏和添加字段。
字段操作为业务操作提供了很大的帮助，这里思想参考了七月老师的编程思想，推荐仔细阅读和学习lin-cms-flask这个优秀的项目。
Jeffrey的博客
# patch/db.py
class SQLAlchemy(_SQLAlchemy):
@contextmanager
def auto_commit(self):
try:
yield
self.session.commit()
except Exception as e:
self.session.rollback()
raise e
class Pagination(_Pagination):
def hide(self, *keys):
for item in self.items:
item.hide(*keys)
return self
def append(self, *keys):
for item in self.items:
item.append(*keys)
return self
class BaseQuery(_BaseQuery):
def filter_by(self, not_del: bool = True, **kwargs):
"""
查询未被软删除的记录
"""
if not_del:
kwargs['delete_time'] = None
return super(BaseQuery, self).filter_by(**kwargs)
def paginate(self, page=1, size=20, error_out=True, max_per_page=None):
"""
覆写分页
"""
if max_per_page is not None:
size = min(size, max_per_page)
items = self.limit(size).offset((page - 1) * size).all()
total = self.order_by(None).count()
return Pagination(self, page, size, total, items)
编写基类是为了让其他数据模型继承该类，简化其他类的编写，id、create_time、update_time、delete_time 这些必须的字段可以写在基类中，id主键可以根据业务更改，可以是自增ID、雪花ID、UUID 等，在数据模型中使用 keys() 方法可以在模型层面对隐藏和添加字段，通过 hide()、append() 在业务层面隐藏和添加字段。
# model/base.py
class BaseModel(db.Model):
__abstract__ = True
id = Column('id', String(36), default=uuid4().hex, primary_key=True, comment='主键标识')
create_time = Column('create_time', DateTime, server_default=func.now(), comment='创建时间')
update_time = Column('update_time', DateTime, onupdate=func.now(), comment='更新时间')
delete_time = Column('delete_time', DateTime, comment='删除时间')
def __getitem__(self, key):
return getattr(self, key)
@orm.reconstructor
def init_on_load(self):
"""
无法直接调用构造函数 需使用装饰器
"""
self._fields = ['status']
self._exclude = ['delete_time', 'update_time']
self.__set_fields()
def __set_fields(self):
columns = inspect(self.__class__).columns
all_columns = set([column.name for column in columns])
self._fields.extend(list(all_columns - set(self._exclude)))
def keys(self):
return self._fields
def hide(self, *keys):
for key in keys:
hasattr(self, key) and self._fields.remove(key)
return self
def append(self, *keys):
for key in keys:
hasattr(self, key) and self._fields.append(key)
return self
@property
def status(self):
return not self.delete_time
@classmethod
def get_or_404(cls, **kwargs):
rv = cls.query.filter_by(**kwargs).first()
if not rv:
raise NotFound
return rv
@classmethod
def all_or_404(cls, **kwargs):
rv = cls.query.filter_by(**kwargs).all()
if not rv:
raise NotFound
return rv
@classmethod
def get_one(cls, **kwargs):
return cls.query.filter_by(**kwargs).first()
@classmethod
def get_all(cls, **kwargs):
return cls.query.filter_by(**kwargs).all()
@classmethod
def create(cls, commit: bool = True, **kwargs):
instance = cls()
for attr, value in kwargs.items():
hasattr(instance, attr) and setattr(instance, attr, value)
return instance.save(commit)
def update(self, commit: bool = True, **kwargs):
for attr, value in kwargs.items():
hasattr(self, attr) and setattr(self, attr, value)
return self.save(commit)
def save(self, commit: bool = True):
db.session.add(self)
commit and db.session.commit()
return self
def delete(self, commit: bool = True, soft: bool = True):
"""
删除 默认使用软删除
"""
if soft:
self.delete_time = func.now()
self.save()
else:
db.session.delete(self)
commit and db.session.commit()
@classmethod
def get_pagination(cls, not_del: bool = True):
"""
分页调用
"""
validator = PaginateValidator().dt_data
page = validator.get('page')
size = validator.get('size')
paginator = cls.query
if not_del:
paginator = paginator.filter_by(delete_time=None)
return paginator.order_by(cls.create_time.desc()).paginate(page=page, size=size)
模型中使用
# model/user.py
def keys(self):
self.hide('password').append('name')
return self._fields
业务中使用
# v1/user.py
@api.route('', methods=['GET'])def get_users():
"""
分页查询用户
"""
pagination = User.get_pagination().append('update_time', 'delete_time')
return Success(data=paginator_schema(pagination))
编写好模型后，需要使用 migrate 生成模型：
flask db init
flask db migrate
flask db upgrade
需要注意的是，如果变更了字段类型或长度时，migrate 默认不会产生新变更，此时需要在 migrations/env.py 的 run_migrations_online 函数中加入：
context.confgure(
# 检查字段类型
compare_type=True,
# 比较默认值
compare_server_default=True,
connection=connection,
target_metadata=target_metadata,
process_revision_directives=process_revision_directives,
**current_app.extensions['migrate'].configure_args
)