Flask了解一下

Posted 2021-04-12 时雨

Flask是Python的轻量级Web后端框架，所谓轻量级——without orm，without templates language，without login manage，without db session manage，without admin...所以你可能要在一些方案中去选择你要的东西.

我把Web后端开发简单分为以下几步：

   0. 数据库表设计；

1. 路由设计；

2. html/JS/CSS开发;

3. 后端表单数据校验及数据函数设计；

4. 视图实现；

0.数据库表设计：）

根据需求设计数据表，定义清楚表字段名、字段类型、字段默认值、字段限制，表与表的关联（字段注释很重要），这里ORM选用SqlAlchemy，会话管理flak-sqlalchemy-session：

from sqlalchemy import String
from sqlalchemy import Column
from sqlalchemy import DateTime
from sqlalchemy import Integer
from commons.funtions import get_now
from db.sqlite_engine import Base
from db.sqlite_engine import engine


class Example(Base):
    __tablename__ = 'example'
    
   STATUS_ACTIVE = 1
   STATUS_BLOCK = 0
   STATUS = {STATUS_ACTIVE: '激活', STATUS_BLOCK: '冻结'}
    
    # id
    id = Column(Integer, primary_key=True)
    # 版本
    version = Column(Integer, default=1)
    # 用户名
    name = Column(String(16), nullable=False)
    # 邮箱
    email = Column(String(64), nullable=False, unique=True)
    # 创建时间
    create_time = Column(DateTime, default=get_now())
    # 状态
    status = Column(Integer, default=STATUS_BLOCK)

基本数据类型Interger\String\DateTime\Text\Boolean\Float，超长文本LONGTEXT可能会用到。

---sqlalchemy操作数据---

查询

# 简单查询
# 查询冻结状态
query = current_session.query(Example).\
    filter_by(status=Example.STATUS_BLOCK)
# 第1条记录
row = query.first()
# 所有记录
rows = query.all()

# 复杂查询
# 邮箱含'shenweiwei'并且状态激活
query = current_session.query(Example).\
    filter(Example.email.contains('shenweiwei')).\
    filter(Example.status == Example.STATUS_ACTIVE)
# 第1条记录
row = query.first()
# 所有记录
rows = query.all()

简单等于查询用filter_by即可，可接入**dict作为参数，filter不可以带入多个参数，属于查询：Example.id.in_(id_list)，不属于：Example.id.notin_(id_list)，不等于：Example.id != Example.STATUS_BLOCK。

添加

# 添加一条记录
example = {'name': 'example', 'email': '[e-mail]'}
row = Example(**example)
current_session.add(row)
current_session.commit()

# 添加多条记录
examples = [Example(name='example1', email='[e-mail]'),
            Example(name='example2', email='[e-mail]')]
current_session.bulk_save_objects(examples)
current_session.commit()

更新

# 更新一条数据
row = current_session.query(Example).\
    filter_by(name='shenweiwei').first()
row.status = Example.STATUS_ACTIVE
current_session.commit()

# 更新多条数据
count = current_session.query(Example).\
    filter(Example.name.contains('shen')).\
    update({'status': Example.STATUS_ACTIVE})

删除

# 删除一条数据
row = current_session.query(Example).\
    filter_by(name='shenweiwei').first()
current_session.delete(row)
current_session.commit()

# 删除多条数据
current_session.query(Example).\
    filter(Example.name.contains('shen')).\
    delete(synchronize_session=False)
current_session.commit()

1.路由设计：）

blueprint（蓝图）是flask用于组织路由/项目结构的模块，类似于Django的app，你可以在app里定义该app的相关路由，blueprint与之相似。当使用blueprint时，路由的标识将会添加上路由所在blueprint名作为前缀：bp_meet.index。

对于C（create）U（update）R（retrieve）D（delete），我们可以定义项目路由规范，例如统一的路由格式module/operate[/id]，operate是（create，update，retrieve，delete）的元素，往往我们需要扩展operates，加入restart，stop，start等一系列操作。下面是定义一条新增应用的路由：

@bp_application.route('/application/add', methods=['POST', 'GET'],
                      endpoint='add')
@login_required
def application_add():
    """应用创建"""
    pass

2.HTML/JS/CSS开发：）

HTML表单内容是由数据库表（数据）和路由（URL）来确定的，表单字段名务必与数据库表字段名保持一直！！！以便表单数据格式化直接添加/更新到数据库，不一致也没关系啦，不炸毛就好。

刷新页面的话，js里form.submit()提交数据：

<script>
    $(document).ready(function () {
        $("#sub-btn").click(function () {
            var form = $('#form');
            form.submit();
        });
    })
</script>

只刷新数据的话，js里ajax放form.serializeArray()提交数据：

$.ajax({
    cache: false,
    type: 'post',
    url: '',
    data: form.serializeArray(),
    async: true,

    success: function (resp) {
        if (resp.status) {
            window.location.href = '/sql/index';
        } else {
            alert(resp.message);
        }
    }
});

3.后端表单数据校验及数据函数设计：）

WTForms作为后端数据校验模块，除了字段内定义的校验器外，还可以定义validate_field(form, field)函数自定义field字段的校验，常用于复杂校验，如数据库查询校验，多字段关联校验。

form.validate() 检查所有字段的校验器，通过返回True，不通过则在form.errors里存放错误信息。

from flask_sqlalchemy_session import current_session
from wtforms import StringField
from wtforms import IntegerField
from wtforms import validators
from wtforms import ValidationError

from commons.base_form import BaseForm
from business.example.tables import Example


class IdVersion(BaseForm):
    """Id Version基类"""
    id = IntegerField('id',
                      [validators.required(message='id is required.')])
    version = IntegerField('version',
                           [validators.required(message='version is required.')])

    def validate_id(self, _):
        """校验id-version"""
        class_ = self.__class__.target
        if current_session.query(class_).\
                filter_by(id=self.id.data,
                          version=self.version.data).first():
            raise ValidationError('record is not exist.')


class ExampleBaseForm(BaseForm):
    """Example基类"""
    name = StringField('name',
                       [validators.input_required(message='name is required.'),
                        validators.length(max=16, message='name_max_length=16')])
    email = StringField('email',
                        [validators.input_required(message='email is required.'),
                         validators.length(max=64, message='email_max_length=64')])
    status = IntegerField('status', [validators.optional()])

    def data_parser(self):
        data = self.data
        data.pop('csrf_token', None)
        return data


class ExampleAddForm(ExampleBaseForm):
    """添加Example"""

    @staticmethod
    def validate_email(_, field):
        """添加时 校验邮箱"""
        row = current_session.query(Example).filter_by(email=field.data).first()
        if row:
            raise ValidationError('user email existed: %s' % field.data)


class ExampleEditForm(ExampleBaseForm, IdVersion):
    """修改Example"""
    target = Example

    @staticmethod
    def validate_email(form, field):
        """修改时 校验邮箱"""
        row = current_session.query(Example).filter_by(email=field.data).first()
        if row and row != form.id.data:
            raise ValidationError('user email existed: %s' % field.data)

上面创建了四个Form类，IdVersion用于所有表单修改/删除数据时，记录版本的校验；ExampleBaseForm用于被新增/修改Form继承使用，ExampleAddForm是新增数据Form，ExampleEditForm是修改Form，ExampleEditForm与ExampleAddForm会在部分字段的校验上有差异，例如上面的validate_email实现是不一样的，删除Form可参考IdVersion。

最近发现Django和Flask在Form处理上有着一些差异，有一点，Django实例化的form具有所有提交的数据字段，不管这些字段在Form类中是否有定义，而Flask实例化的form只具有Form中类的字段。Django的这种“随和”性质让一些人的代码越加恣意妄为：新加字段不写到Form类里，不校验，反正不会报错...

往往由表单提交到后端的数据不能直接保存到数据库，需要对数据再处理，比如添加操作人，这时，我会给form类添加一个data_parser()的函数，专门处理数据，返回供新增/更新数据库记录的dict数据：

class ExampleBaseForm(BaseForm):
    """校验"""
    pass
    
def data_parser(self):
    data = self.data
    data.pop('csrf_token', None)
    return data        

4.视图实现：）

视图函数是最后一步，可能有些新人上来就是先写这个，其实不然，0到3是基石，当然第4步是最重要的业务实现部分，这里只是简单的介绍了Flask下的web开发过程，以供参考。

@bp_application.route('/application/add', methods=['POST', 'GET'],
                      endpoint='add')
@login_required
def application_add():
    """应用创建"""
    if request.method == 'GET':
        return render_template('/application/application_add.html', **locals())
    form = ApplicationAddForm()
    if not form.validate():
        return jsonify({'status': False, 'errors': form.errors.values()})
    app = form.data_parser()
    row = Application(**app)
    current_session.add(row)
    current_session.commit()
    return jsonify({'status': True, 'message': 'add success.'})

一次流畅的开发是0到4一气呵成，而不是在4的时候不断修改0到3.

references:

flask http://docs.jinkan.org/docs/flask/tutorial/index.html#tutorial

sqlalchemy http://docs.sqlalchemy.org/en/latest/core/tutorial.html

flask_sqlalchemy_session http://flask-sqlalchemy-session.readthedocs.io/en/v1.1/

wtforms https://wtforms.readthedocs.io/en/stable/#

flask-wtf http://flask-wtf.readthedocs.io/en/stable/quickstart.html