Jeffrey的博客 Jeffrey的博客

左脑编程,右脑写诗

目录
Flask构建Web API Wheel(六)—— 数据库操作基类
/      

Flask构建Web API Wheel(六)—— 数据库操作基类

Flask-SQLAlchemy操作数据库十分的方便,但也不影响我们再为它封装一个数据库操作基类,

覆写SQLALchemy

patch包是用于存放被覆写的模块,在其中新建db.py,分别覆写SQLAlchemy、Pagination、BaseQuery,使得其支持数据库操作自动提交,查询未被软删的记录,以及在业务上隐藏和添加字段,特别像Django序列号器的隐藏和添加字段。

字段操作为业务操作提供了很大的帮助,这里思想参考了七月老师的编程思想,推荐仔细阅读和学习lin-cms-flask这个优秀的项目。

# 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
)

标题:Flask构建Web API Wheel(六)—— 数据库操作基类
作者:Jeffrey