FlaskWeb开发实战2_实战

第七章留言板

使用click创建虚拟数据

@app.cli.command()
@click.option('--count', default=20, help='Quantity of messages, default is 20.')
def forge(count):
    """Generate fake messages."""
    from faker import Faker

    db.drop_all()
    db.create_all()

    fake = Faker()
    click.echo('Working...')

    for i in range(count):
        message = Message(
            name=fake.name(),
            body=fake.sentence(),
            timestamp=fake.date_time_this_year()
        )
        db.session.add(message)

    db.session.commit()
    click.echo('Created %d fake messages.' % count)

Bootstrap-Flask 插件

注意 , 不是flask-Boostrap.

1	pip install bootstrap-flask

Bootstrap-Flask插件最牛逼的地方就是内置了很多常用的宏

宏	所在模板路径	说明
render_field	bootstrap/form.html	渲染单个 WTForms表单字段
render_form	bootstrap/form.html	渲染整个 WTForms表单类
render_pager	bootstrap/pagination.html	渲染一个基础分页导航，仅包含上一页、下一页按钮
render_pagination	bootstrap/pagination.html	渲染一个标准分页导航部件
render_nav_item	bootstrap/nav.html	渲染导航链接
render_breadcrumb_item	bootstrap/nav.html	渲染面包屑链接

Flask-Moment 本地化日期和时间

Moment.js是一个用于处理时间和日期的开源JavaScript库，它可以对时间和日期进行各种方式的处理。
它会根据用户电脑中的时区设置在客户端使用 Javascript来渲染时间和日期，
另外它还提供了丰富的时间渲染格式支持。

1	pip install flask-moment

使用datetime.utcnow这个纯正时间替代 datetime.now这个细致时间

class Message(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(20))
    body = db.Column(db.String(200))
    timestamp = db.Column(db.DateTime, default=datetime.utcnow, index=True)

第八章个人博客

为什么使用Endpoint?

我们在使用url_for()的时候 , 定位的都是Endpoint , 而不是使用视图函数 . 为啥?

因为使用端点可以实现蓝本的视图函数命名空间。即“蓝本名.视图函数名” , 这样就支持视图函数重名了

邻接列表关系

楼中楼功能 : 给某条回复添加新的回复
方法:
- 方法一 : 创建Cormment和Replay表 , 然后使用一对多关系将评论和回复关联起来。
- 方法二 : 因为回复本身也是评论，如果可以在评论模型内建立层级关系，那么就可以在一个模型中表示评论和回复。
这种在同一个模型内的一对多关系在 SQLALchemy中被称为邻接列表关系

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(60))
    body = db.Column(db.Text)
    timestamp = db.Column(db.DateTime, default=datetime.utcnow, index=True)
    comments = db.relationship('Comment', back_populates='post', cascade='all, delete-orphan')

class Comment(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    author = db.Column(db.String(30))
    body = db.Column(db.Text)

    replied_id = db.Column(db.Integer, db.ForeignKey('comment.id'))
    post_id = db.Column(db.Integer, db.ForeignKey('post.id'))

    post = db.relationship('Post', back_populates='comments')
    # 子回复
    replies = db.relationship('Comment', back_populates='replied', cascade='all, delete-orphan') 
    # 父回复
    replied = db.relationship('Comment', back_populates='replies', remote_side=[id])

在这个关系函数中，通过将 remote side参数设为id字段，我们就把id字段定义为关系的远程侧（Remote side），而 replied id就相应地变为本地侧（Local side），这样反向关系就被定义为多对一，即多个回复对应一个父评论。

渲染导航链接

因为请求是先经过View函数 , 此时已经有endpoint
所以可以使用request.endpoint== 'blog.index' 来确定激活的sidebar

1
2
3

<li {% request.endpoint == 'blog.index' %}  class="active" {% endif %}>
	<a href="{{ url_for('blog.index''') }}">Home</a>
</li>

切换博客主题(切换CSS)

切换博客主题 , 其实就是切换CSS文件 , 一般的操作思路如下

定义一个view专门来切换cookies . 前端按钮触发这个View

1	BLUELOG_THEMES = {'perfect_blue': 'Perfect Blue', 'black_swan': 'Black Swan'}

@blog_bp.route('/change-theme/<theme_name>')
def change_theme(theme_name):
    if theme_name not in current_app.config['BLUELOG_THEMES'].keys():
        abort(404)

    response = make_response(redirect(url_for('blog.index')
    response.set_cookie('theme', theme_name, max_age=30 * 24 * 60 * 60)
    return response

修改template

1	<link rel="stylesheet" href="{{ url_for('static', filename='css/%s.min.css' % request.cookies.get('theme', 'perfect_blue')) }}" type="text/css">

主题列表 :

 <div class="dropdown-menu" aria-labelledby="dropdownMenuButton">
 {% for theme_name, display_name in config.BLUELOG_THEMES.items() %}
 	<a class="dropdown-item" href="{{ url_for('blog.change_theme', theme_name=theme_name, next=request.full_path) }}"> {{ display_name }}</a>
{% endfor %}
 </div>

调用User模型的get_id()函数 (该函数返回User的唯一标识 , 如果User继承UserMixin , 那么函数get_id返回primaryKey)
将get_id()的结果存入session
每次执行被@login_required装饰的View的时候 , 都会从session中拿到这个唯一标识 . 接着把这个唯一标识传入load_user函数(该函数就是根据唯一标识找到对应的User实例)
login user的remember参数 : 设为True时 Flask-Login会在用户浏览器中创建一个名为 remember token的 cookie，当通过 session设置的 user_id cookie因为用户关闭浏览器而失效时，它会重新恢复 user_id cookie的值。

currenr_user

currenr_user是一个和 current app类似的代理对象（Proxy），表示当前用户。调用时会返回与当前用户对应的用户模型类对象。
因为session中只会存储登录用户的id，所以为了让它返回对应的用户对象，我们还需要设置一个用户加载函数。这个函数需要使用 login manager.user_loader装饰器，它接收用户id作为参数，返回对应的用户对象，
1
2
3
@loginmanager.user_loader
def load_user(user_id):
return User.query.filter_by(id=user_id).first()

为一整个蓝图添加login_required

为蓝本注册一个 before request处理函数，然后为这个函数附加 login required装饰器。
因为使用 before request钩子注册的函数会在每一个请求前运行，所以这样就可以为该蓝本下所有的视图函数添加保护，函数内容可以为空

@bp.brefor_request
@login_required
def login_protest():
    pass

使用CSRFProtect拓展手动实现CSRF防御

CSRFProtect类内置于WTForm , 因此可以直接使用

from flask_wtf import CSRFProtect
csrf = CSRFProtect()

csrf.init_app(app)

创建一个hidden的input字段 ,name为csrf_token , value为{{ csrf_token() }}

<form class="inline" method="post"
      action="{{ url_for('admin.delete_post', post_id=post.id, next=url_for('blog.index')) }}">
    <input type="hidden" name="csrf_token" value="{{ csrf_token() }}"/>
    <button type="submit" class="btn btn-danger btn-sm" onclick="return confirm('Are you sure?');">Delete
    </button>
</form>

CSRFProtect 会自动获取并验证csrf令牌
默认情况下，当令牌验证出错或过期时，程序会返回400错误，也可以单独创建一个错误处理函数捕捉令牌出错时抛出的 CSRFError异常，
Werkzeug内置的HTTP异常类 , 都会将错误描述保存在异常对象的description属性中。

from flask_wtf.csrf import CSRFError

def register_errors(app):
    @app.errorhandler(404)
    def page_not_found(e):
        return render_template('errors/404.html'), 404
    
    @app.errorhandler(CSRFError)
    def handle_csrf_error(e):
        # Werkzeug内置的HTTP异常类 , 都会将错误描述保存在异常对象的description属性中。
        return render_template('errors/400.html', description=e.description), 400

取消对某个蓝图的CSRF防御

from flask_wtf import CSRFProtect

api_v1 = Blueprint('api_v1', __name__)
csrf = CSRFProtect()
csrf.exempt(api_v1)

第九章图片社交网站

项目组织架构

在大型 Flask项目中，主要有三种常见的项目组织架构：

功能式架构

在功能式架构中，程序包由各个代表程序组件（功能）的子包组成，比如 blueprints（蓝本）forms（表单）、templates（模板）、models（模型）等，在这些子包中，按照程序的板块分模块来组织代码，比如 forms子包下包含 front.py、auth,py和 dashboard py
分区式架构

在分区式架构中，程序被按照自身的板块分成不同的子包。myapp使用分区式架构可以分别创建 front、auth和 dashboard三个子包，这些子包直接在程序包的根目录下创建，子包中使用模块组织不同的程序组件，比如 vIews.py、forms。py等。这种分类自然决定了每一个子包都对应着一个蓝本，这时蓝本在每个子包的构造文件中创建。
混合式架构

混合式架构，顾名思义，就是不按照常规分类来组织。比如，采用类似分区式架构的子包来组织程序，但各个蓝本共用程序包根目录下的模板文件夹和静态文件文件夹

在线的占位图片服务

1 2	{# id这个query-args是为了破环存缓 #} <img src="https://picsum.photos/800/?random&id={{ photo .id }} " >

URL类似https://picsum.photos/600/800.如果想获取正方形的图片，那么只传递一个尺寸数字就可以了。
如果你想要每次请求都获得随机的图片，可以在URL后面附加?random。
不过这会有一点问题。因为当浏览器发现你有多个发往同一个URL的请求时，它会使用缓存的响应，这样你的图片就不再是随机的了。为了避免浏览器这个“好心”的缓存行为，我们可以在URL后附加一个无意义的查询字符串，使用数据库图片记录的id填充。这个查询字符串会被服务器忽略，但因为每个图片URL的参数都不同，浏览器会把它们都当作不同的请求来处理，
这种技术被称为 Cache Busting（缓存破坏）。

WTForm的EqualTo和Regexp校验器

class RegisterForm(FlaskForm):
    username = StringField('Username', validators=[DataRequired(), Length(1, 20),Regexp('^[a-zA-Z0-9]*$')])
    password = PasswordField('Password', validators=[DataRequired(), Length(8, 128), EqualTo('password2')])
    password2 = PasswordField('Confirm password', validators=[DataRequired()])

生成JWS（json Web Signature）

JWS由三部分组成，它们分别是存储签名所使用的算法、签名时间和过期时间的头部（Header）、存储数据的负载（Payload）和签名（Signature）。
序列化对象提供一个 dumps方法来写入数据，它接收包含数据的字典对象作为参数。它会根据过期时间创建头部（Header），然后将数据编码到JWS的负载（Payload）中，再使用密钥对令牌进行签名（Signature），最后将签名序列化后生成令牌值。
简单来说 , 只要为TimedJSONWebSignatureSerializer的dumps()函数传入query_args , 即可生成token
后端接受到token后 , 调用TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY']).loads(token) , 如果没有报错 , 即可证明验证通过
之后就可以通过data.get(‘operation’)来获取query_args

from itsdangerous import TimedJSONWebSignatureSerializer as Serializer

def generate_token(user, operation, expire_in=None, **kwargs):
    s = Serializer(current_app.config['SECRET_KEY'], expire_in)
    data = {'id': user.id, 'operation': operation}
    data.update(**kwargs)
    # 根据data里的参数 , 生成token
    return s.dumps(data)

token = generate_token(user=user, operation='confirm')

验证token :

def validate_token(user, token, operation, new_password=None):
    try:
        data = Serializer(current_app.config['SECRET_KEY']).loads(token)
    except (SignatureExpired, BadSignature):
        return False

    if operation != data.get('operation') or user.id != data.get('id'):
        return False
    if operation == 'confirm':
        user.confirmed = True
    elif operation == 'reset-password':
        user.set_password(new_password)
    elif operation == 'change-email':
        new_email = data.get('new_email')
        if new_email is None:
            return False
        if User.query.filter_by(email=new_email).first() is not None:
            return False
        user.email = new_email
    else:
        return False

    db.session.commit()
    return True

@auth_bp.route('/confirm/<token>')
@login_required
def confirm(token):
    if current_user.confirmed:
        return redirect(url_for('main.index'))

    if validate_token(user=current_user, token=token, operation='confirm'):
        flash('Account confirmed.', 'success')
        return redirect(url_for('main.index'))
    else:
        flash('Invalid or expired token.', 'danger')
        return redirect(url_for('.resend_confirm_email'))

flask的装饰器必须使用wraps装饰器

使用 functors模块提供的 wraps装饰器可以避免被装饰函数的特殊属性被更改，比如函数名称__name__被更改。
如果不使用该模块，则会导致函数名称被替换，从而导致端点（端点的默认值即函数名）出错

flash函数中返回危险文本的操作

flash函数发送的消息在模板中渲染，消息内容会被自动转义为普通文本。
可以在模板中使用safe过滤器来避免Jina2对变量转义，但对fash消息使用safe过滤器会造成安全隐患，因为攻击者可能会篡改消息内容。
更安全的做法是将传入flash函数的文本转换为 Markup对象。Flask提供的 Markup类可以将文本标记为安全文本，从而避免在渲染时对Jina2进行转义。

message = Markup(
    'Please confirm your account first.'
    'Not receive the email?'
    '<a class="alert-link" href="%s">Resend Confirm Email</a>' %
    url_for('auth.resend_confirm_email'))
flash(message, 'warning')

基于用户角色的权限管理 RBAC (Role-Based Access Control)

角色和权限model设计

角色与权限模型

roles_permissions = db.Table('roles_permissions',
                             db.Column('role_id', db.Integer, db.ForeignKey('role.id')),
                             db.Column('permission_id', db.Integer, db.ForeignKey('permission.id'))
                             )

class Permission(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(30), unique=True)
    roles = db.relationship('Role', secondary=roles_permissions, back_populates='permissions')


class Role(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(30), unique=True)
    users = db.relationship('User', back_populates='role')
    permissions = db.relationship('Permission', secondary=roles_permissions, back_populates='roles')

class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(20), unique=True, index=True)
    password_hash = db.Column(db.String(128))
    role_id = db.Column(db.Integer, db.ForeignKey('role.id'))
    role = db.relationship('Role', back_populates='users')

权限定义示例

操作	权限名称	说明
关注用户	FOLLOW	关注其他用户
收藏图片	COLLECT	添加图片到自己的收藏
发表评论	COMMENT	在图片下添加评论
上传图片	UPLOAD	上传图片
协管员权限	MODERATE	管理资源权限，可以管理网站的用户、图片、评论、标签等资源
管理员权限	ADMINISTER	管理用户角色、编辑网站信息等

角色定义示例

角色名称	拥有的权限	说明
访客（Guest）	仅可以浏览页面	未登录用户
被封禁用户（Blocked）	仅可以浏览页面	因违规行为被封禁账号，禁止登录的用户
被锁定用户（Locked）	FOLLOW、COLLECT	因违规行为被锁定的用户
普通用户（User）	FOLLOW、COLLECT、COMMENT、UPLOAD	注册后用户获得的默认角色
协管员（Moderator）	除了拥有普通用户具有的权限外，还拥有管理网站拥有 MODERATE权限	除了普通用户的权限外内容的权限，负责网站内容管理和维护
管理员（Administrator）	除了拥有普通用户和协管员的所有权限外，还拥有 ADMINISTER权限	拥有所有权限的网站管理员

roles_permissions_map = {
    'Locked': ['FOLLOW', 'COLLECT'],
    'User': ['FOLLOW', 'COLLECT', 'COMMENT', 'UPLOAD'],
    'Moderator': ['FOLLOW', 'COLLECT', 'COMMENT', 'UPLOAD', 'MODERATE'],
    'Administrator': ['FOLLOW', 'COLLECT', 'COMMENT', 'UPLOAD', 'MODERATE', 'ADMINISTER']
}

这里只有四种角色，没有访客和被封禁用户。

访客不需要写入数据库，因为访客的作用就是用来表示不在数据库中的用户。
而被封禁的用户不允许登录，虽然这类用户拥有账户，但是其权限状态和访客完全相同。

填充角色权限表数据 :

class Role(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(30), unique=True)
    users = db.relationship('User', back_populates='role')
    permissions = db.relationship('Permission', secondary=roles_permissions, back_populates='roles')

    @staticmethod
    def init_role():
        roles_permissions_map = {
            'Locked': ['FOLLOW', 'COLLECT'],
            'User': ['FOLLOW', 'COLLECT', 'COMMENT', 'UPLOAD'],
            'Moderator': ['FOLLOW', 'COLLECT', 'COMMENT', 'UPLOAD', 'MODERATE'],
            'Administrator': ['FOLLOW', 'COLLECT', 'COMMENT', 'UPLOAD', 'MODERATE', 'ADMINISTER']
        }

        for role_name in roles_permissions_map:
            role = Role.query.filter_by(name=role_name).first()
            if role is None:
                role = Role(name=role_name)
                db.session.add(role)
            role.permissions = []
            for permission_name in roles_permissions_map[role_name]:
                permission = Permission.query.filter_by(name=permission_name).first()
                if permission is None:
                    permission = Permission(name=permission_name)
                    db.session.add(permission)
                role.permissions.append(permission)
        db.session.commit()

替换AnonymousUserMixin

class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(20), unique=True, index=True)
    role = db.relationship('Role', back_populates='users')
    @property
    def is_admin(self):
        return self.role.name == 'Administrator'

User模型如上 , 现在我们就可以使用current_user来代理User模型了 , 所以我们可以使用current_user.is_admin来判断是不是管理员
但是 , 如果current_user是AnonymousUserMixin的话就不行了 , 因为AnonymousUserMixin对象没有is_admin属性

此时的方法就是替换AnonymousUserMixin

from flask_login import LoginManager, AnonymousUserMixin

login_manager = LoginManager()

class Guest(AnonymousUserMixin):
    @property
    def is_admin(self):
        return False

login_manager.anonymous_user = Guest

一般这个操作放到extensions 模块中创建

权限装饰器及装饰器的继承(装饰器里调用装饰器)

def permission_required(permission_name):
    def decorator(func):
        @wraps(func)
        def decorated_function(*args, **kwargs):
            if not current_user.can(permission_name):
                abort(403)
            return func(*args, **kwargs)
        return decorated_function
    return decorator

# 继承permission_required装饰器 , 生成新的admin_required装饰器
def admin_required(func):
    return permission_required('ADMINISTER')(func)

def moderator_required(func):
    return permission_required('Moderator')(func)

Flask-Dropzone 优化文件上传

Dropzone. js是一款优秀的文件上传插件

详情参考 : http://greyli.com/flask-dropzone/ , https://www.dropzonejs.com/

1	pip install flask-dropzone

1 2	from flask_dropzone import Dropzone dropzone = Dropzone()

重要配置

设置文件类型过滤

Flask-Dropzone内置了一些文件类型（通过MIME定义），可选的值和对应的文件类型如下所示：

default：默认值，运行所有类型
image：图片
audio：音频
video：视频
text：文本
app：程序

1	app.config['DROPZONE_ALLOWED_FILE_TYPE'] = 'image'

如果你想要自己定义允许的文件类型列表，那么你需要将DROPZONE_ALLOWED_FILE_CUSTOM设置True，然后传入一个包含允许的文件后缀名列表组成的字符串给DROPZONE_ALLOWED_FILE_TYPE变量，使用逗号分隔多个后缀名，

1 2	app.config['DROPZONE_ALLOWED_FILE_CUSTOM'] = True app.config['DROPZONE_ALLOWED_FILE_TYPE'] = 'image/*, .pdf, .txt'

# 上传字段的name值 , 默认file
DROPZONE_INPUT_NAME

# 一次可以上传的文件数量最大值 , 默认null’
DROPZONE_MAX_FILES

# 是否在单个请求中发送多个文件，默认一个请求发送一个文件(False)
DROPZONE_UPLOAD_MULTIPL

# 是否开启CSRF保护 , 默认False
DROPZONE_ENABLE_CSRF

# 上传完成后重定向的目标端点
DROPZONE REDIRECT VIEW

简单示例

@main_bp.route('/upload', methods=['GET', 'POST'])
@permission_required('UPLOAD')
def upload():
    if request.method == 'POST' and 'file' in request.files:
        f = request.files.get('file')
        filename = rename_image(f.filename)
        f.save(os.path.join(current_app.config['ALBUMY_UPLOAD_PATH'], filename))
    return render_template('main/upload.html')

{% block styles %}
    {{ super() }}
    <link rel="stylesheet" href="{{ url_for('static', filename='css/dropzone.min.css') }}" type="text/css">
    {{ dropzone.style('margin: 20px 0; border: 2px dashed #0087F7; min-height: 400px;') }}
{% endblock %}

{% block content %}
    <div class="row">
        <div class="col-md-12">
            {{ dropzone.create(action='main.upload') }}
            <a class="btn btn-light float-right" href="{{ url_for('user.index', username=current_user.username) }}">
                Done
            </a>
        </div>
    </div>
{% endblock %}

{% block scripts %}
    {{ super() }}
    <script src="{{ url_for('static', filename='js/dropzone.min.js') }}"></script>
    {{ dropzone.config() }}
{% endblock %}

Flask-Avatars 处理用户头像

1	pip install flask-avatars

文件删除

按照一般的做法，在数据库中删除某张图片前，首先要在文件系统中删除对应的图片文件，但是有更好的实现方法
为Photo模型创建一个数据库事件监听函数。这个监听函数的作用是，当 Photo记录被删除时，自动删除对应的文件。

class Photo(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    description = db.Column(db.String(500))
    filename = db.Column(db.String(64))
    filename_s = db.Column(db.String(64))
    filename_m = db.Column(db.String(64))
    
@db.event.listens_for(Photo, 'after_delete', named=True)
def delete_photos(**kwargs):
    target = kwargs['target']
    for filename in [target.filename, target.filename_s, target.filename_m]:
        path = os.path.join(current_app.config['ALBUMY_UPLOAD_PATH'], filename)
        if os.path.exists(path):  # not every filename map a unique file
            os.remove(path)

当页编辑的最佳实践

比如图片展示列表 , 点击时当页修改图片描述
每个项目下面都渲染一个form , 但是form设置为display:none , 点击时show .
注意这个form应该添加一个取消button , 点击这个button时 , 重新让这个formdisplay:none

<div id="description-form">
    <form action="{{ url_for('.edit_description', photo_id=photo.id) }}" method="post">
    {{ description_form.csrf_token }}
    {{ render_field(description_form.description) }}
    <a class="btn btn-light btn-sm" id="cancel-description">Cancel</a>
    {{ render_field(description_form.submit, class='btn btn-success btn-sm') }}
    </form>
</div>

<script>
    $("#description-btn").click(function(){
        $("#description").hide();
        $("#description-form").show();
    })
    $("#cancel-description").click(function(){
        $("#description-form").hide();
        $("#description").show();
    })
</script>

注册全局Ajax错误

$(document).ajaxError(function (event, request, settings) {
    var message = null;
    if (request.responseJSON && request.responseJSON.hasOwnProperty('message')) {
        message = request.responseJSON.message;
    } else if (request.responseText) {
        var IS_JSON = true;
        try {
            var data = JSON.parse(request.responseText);
        }
        catch (err) {
            IS_JSON = false;
        }
        if (IS_JSON && data !== undefined && data.hasOwnProperty('message')) {
            message = JSON.parse(request.responseText).message;
        } else {
            message = default_error_message;
        }
    } else {
        message = default_error_message;
    }
    toast(message, 'error');
});

from flask_login import fresh_login_required, login_fresh

class ChangePasswordForm(FlaskForm):
    old_password = PasswordField('Old Password', validators=[DataRequired()])
    password = PasswordField('New Password', validators=[
        DataRequired(), Length(8, 128), EqualTo('password2')])
    password2 = PasswordField('Confirm Password', validators=[DataRequired()])
    submit = SubmitField()

@user_bp.route('/settings/change-password', methods=['GET', 'POST'])
@fresh_login_required
def change_password():
    form = ChangePasswordForm()
    if form.validate_on_submit():
        if current_user.validate_password(form.old_password.data):
            current_user.set_password(form.password.data)
            db.session.commit()
            flash('Password updated.', 'success')
            return redirect(url_for('.index', username=current_user.username))
        else:
            flash('Old password is incorrect.', 'warning')
    return render_template('user/settings/change_password.html', form=form)

fresh_login _required装饰器 : 确保用户处于“活跃”的认证状态。
当用户登录账户时用户会话会被标记为“新鲜的（fresh）”，通过使用 session对象写人名为 fresh的 cookie实现。如果用户会话被销毁或过期了，而用户勾选了“记住我”选项。尽管这时用户仍然保持登录状态，但会话已经被标记为“不新鲜”。出于安全考虑，像修改密码这类敏感操作，应该在“新鲜”会话下进行。

编写一个重新验证视图函数

@auth_bp.route('/re-authenticate', methods=['GET', 'POST'])
@login_required
def re_authenticate():
    if login_fresh():
        return redirect(url_for('main.index'))

    form = LoginForm()
    if form.validate_on_submit() and current_user.validate_password(form.password.data):
        confirm_login()
        return redirect_back()
    return render_template('auth/login.html', form=form)

login_manager.refresh_view ='auth re_authenticate'
login_manager.needs_refresh_message_category='warning'
login_manager.needs_refresh_message='为了保护你的账户安全，请重新登录。'

subquery 与 in_ 与 join

以前的做法 :

query = db.session.query(Follow.followed_id).filter(Follow.follower_id == current_user.id)
followed_list = [f.followed_id for f in query.all()]

followed_photos = Photo.query.filter(Photo.auther_id.in_(followed_list)).order_by(Photo.timestamp.desc()).all()

直接使用subquery , 不必将其改成list

1
2
3

followed_ids = db.session.query(Follow.followed_id).filter(Follow.follower_id == current_user.id).subquery()

followed_photos = Photo.query.filter(Photo.auther_id.in_(followed_ids)).order_by(Photo.timestamp.desc()).all()

简单来说 , subquery的作用就是将query改成table

实际上 , in_ 操作的性能没有join好 , 最好使用join

1	followed_photos = Photo.query.join(Follow, Follow.followed_id == Photo.author_id).filter(Follow.follower_id == self.id)

sqlalchemy随机获取内容

SQLALchemy通过 sqlalchemy.sql.expression模块的func属性提供了泛型函数（Generic Function）支持，使用这个属性可以调用对应的SQL函数。
func.random() : 调用数据库引擎提供的随机函数

1
2
3

from sqlalchemy.sql.expression import func

photos = Photo.query.order_by(func.random()).limit(12)

Flask Whooshee 实现全文搜索

全文搜索的原理是索引程序通过扫描数据库中的每一个词，对每一个词建立一个索引指明该词在数据库中出现的次数和位置，当用户查询时，检索程序通过索引进行查找，并返回匹配的数据。

1	pip instaoll flask-whooshee

Flask-Whooshes默认在项目的根目录下建立名为 whooshes的索引文件夹，你可以通过配置变量 WHOOSHEE_DIR自定义这个位置。
使用 Flask-Whooshee提供的 whooshes.register_model()装饰器，即可对要索引的字段进行注册。
下面我们分别在这三个模型类前附加这个装饰器，传入要被索引的目标字段名称作为参数：

@whooshee.register_model('name', 'username')
class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(20), unique=True, index=True)
    password_hash = db.Column(db.String(128))
    name = db.Column(db.String(30))

默认情况下，Flask-Whooshes在对这几个注册的字段执行写入操作后自动创建并更新索引。在下面这两种情况下，你需要重新生成索引：

索引数据丢失，比如误删除索引文件夹；

在新的程序中使用扩展，这时数据库已经包含许多数据。

最后执行reindex方法来重新创建索引：

1
2
3

flask shell 
>>> from album.extensions import whooshee
>>> whooshe.reindex()

所以我们可以编写一个Celery程序来执行whooshe.reindex()

示例

搜索请求通常会通过GET方法发出，这样可以支持用户对搜索URL进行收藏或分享，同时也可以避免浏览器对于POST请求重复提交问题。
Fask-Whooshes覆盖了 SQALchemy的 query对象，为其添加了一个 whooshes_search()查询方法，这个方法接收搜索关键字作为参数，返回包含所有匹配记录的查询对象。

@main_bp.route('/search')
def search():
    q = request.args.get('q', '').strip()
    if q == '':
        flash('Enter keyword about photo, user or tag.', 'warning')
        return redirect_back()

    category = request.args.get('category', 'photo')
    page = request.args.get('page', 1, type=int)
    per_page = current_app.config['ALBUMY_SEARCH_RESULT_PER_PAGE']
    if category == 'user':
        pagination = User.query.whooshee_search(q).paginate(page, per_page)
    elif category == 'tag':
        pagination = Tag.query.whooshee_search(q).paginate(page, per_page)
    else:
        pagination = Photo.query.whooshee_search(q).paginate(page, per_page)
    results = pagination.items
    return render_template('main/search.html', q=q, results=results, pagination=pagination, category=category)

<form class="form-inline my-2 my-lg-0" action="{{ url_for('main.search') }}">
    <input type="text" name="q" class="form-control mr-sm-1" placeholder="Photo, tag or user"
           required>
    <button class="btn btn-light my-2 my-sm-0" type="submit">
        <span class="oi oi-magnifying-glass"></span>
    </button>
</form>

第十章代办事项程序

单页面应用布局

单页程序，并不意味着我们只能使用一个页面。程序中实际包含三个页面：介绍页、登录页和程序页，这三个页面分别在局部模板 _intro.html、_login.html和 _app.html中定义。
三个模板不会直接加载，而是通过AJAX请求获取并动态插到根页面中。
为了便于在AJAX中发送请求到对应的URL，我们在根页面中定义了多个 JavaScript变量，分别存储指向这三个页面的URL。另外，我们程序中的所有操作都通过 JavaScript发送AJAX请求实现，像登录、注册、注销等这类不包含URL变量的URL，也在这里定义：

{# base.html #}
<!DOCTYPE html>
<html>
<head>
    <meta name="viewport" content="width=device-width, initial-scale=1.0"/>
    <meta charset="utf-8"/>
    <title>Todoism</title>
    <link href="{{ url_for('static', filename='css/materialize.min.css') }}" rel="stylesheet">
    <link href="{{ url_for('static', filename='css/style.css') }}" rel="stylesheet">
    <link rel="icon" href="{{ url_for('static', filename='favicon.ico') }}" type="image/x-icon">
    <link href="http://fonts.googleapis.com/icon?family=Material+Icons" rel="stylesheet">
</head>
<body>
<div id="main"></div>
<script src="{{ url_for('static', filename='js/jquery.min.js') }}"></script>
<script src="{{ url_for('static', filename='js/materialize.min.js') }}"></script>
<script src="{{ url_for('static', filename='js/script.js') }}" type="text/javascript"></script>
<script type="text/javascript">
    var csrf_token = "{{ csrf_token() }}";
    var login_page_url = "{{ url_for('auth.login') }}";
    var app_page_url = "{{ url_for('todo.app') }}";
    var intro_page_url = "{{ url_for('home.intro') }}";
    var new_item_url = "{{ url_for('todo.new_item') }}";
    var clear_item_url = "{{ url_for('todo.clear_items') }}";
    var login_url = "{{ url_for('auth.login') }}";
    var register_url = "{{ url_for('auth.register') }}";
    var logout_url = "{{ url_for('auth.logout') }}";
    var default_error_message = "{{ _('Oops, something was wrong!') }}";
    var empty_body_error_message = "{{ _('Your todo was empty!') }}";
    var login_error_message = "{{ _('Empty username or password!') }}";
</script>
</body>
</html>

在根页面内切换子页面

在单页面应用中 , Flask的视图函数的功能退化为提供数据的内部接口，而真正的视图处理则转移到了客户端JavaScript 中
在单页面中切换子页面时，我们希望这些子页面变化时也产生一个可以被保存为书签并且添加到浏览器历史的URL，而且使浏览器的前进、后退按钮发挥作用。为了让程序的状态在URL中表现出来，我们通过在URL后面添加hash（即URL中#后面的部分）来记录状态。程序中的三个主要页面使用对应的hash标签表示，比如导航栏上的登录按钮
1
<a class="waves-effect waves-light btn red" href="#login">{{ _('Login') }}</a>
单击这个按钮，会访问#login产生的URL类似http://example.com#login表示登录页面。

在URL中添加hash不会产生请求，而我们可以通过监听hash的变化来设置回调函数来更新页面。我们创建一个
hashchange事件的监听函数，用于在hash值改变时执行对应的函数

$(window).bind('hashchange', function () {
    // 有些浏览器不返回#，这里统一去掉#
    var hash = window.location.hash.replace('#', '');
    var url = null;
    // 根据hash值的不同，选择对应的页面URL
    if (hash === 'login') {
        url = login_page_url
    } else if (hash === 'app') {
        url = app_page_url
    } else {
        url = intro_page_url
    }
    // 向对应的页面URL发送GET请求，服务器端会返回对应的局部模板
    $.ajax({
        type: 'GET',
        url: url,
        success: function (data) {
            $('#main').hide().html(data).fadeIn(800); // 插入子页面
            activeM(); // 激活新插入的页面中的 Materia1ize组件
        } // 错误回调已经统一设置，不需要定义 error回调
    });
});

if (window.location.hash === '') {
    window.location.hash = '#intro'; // home page, show the default view
} else {
    $(window).trigger('hashchange'); // 触发 hashchange事件，重新加载页面
}

所以 , 当我们执行其他操作后需要切换页面时，通过切换URL中的hash（通过 window,locationhash）即可切换页面。比如，当用户单击登录按钮时，如果验证通过，我们在发送AJAX请求的success回调函数中将hash设为对应程序页面的App，就可以切换到App页面：
1
2
3
seccess: function (data) {
window.location.hash = "#app";
}

后端代码

@todo_bp.route('/items/new', methods=['POST'])
@login_required
def new_item():
    data = request.get_json()
    if data is None or data['body'].strip() == '':
        return jsonify(message=_('Invalid item body.')), 400
    item = Item(body=data['body'], author=current_user._get_current_object())
    db.session.add(item)
    db.session.commit()
    return jsonify(html=render_template('_item.html', item=item), message='+1')

template的True,False转为true,false

done属性后使用了 Flask内置的 tojson过滤器，因为 Python中的布尔值和 Javascript 中的布尔值（全小写形式的true和 false）不同，所以需要使用 tojson过滤器将变量值转换为JSON格式。

1 2	<div class="row item card-panel hoverable" data-href="{{ url_for('.edit_item', item_id=item.id) }}" data-id="{{ item.id }}" data-body="{{ item.body }}" data-done={{ item.done\|tojson }}>

使用 Flask-Babel 支持国际化与本地化

国际化：国际化指设计和修改程序以便让程序支持多种语言或区域，而不是固定于某个语言或区域。国际化为本地化做了程序上的准备。
本地化：本地化指为程序添加某些资源（比如翻译文件）以便支持某个特定的语言或区域。本地化通常会进行多次，比如要支持10种语言，那么就要进行10次本地化处理。

1	pip install flask-babel

区域和语言

我们使用语言代码（language code）和区域代码（locale code）来区分不同的区域和语言。
语言代码 : 我们最常见到的语言代码有zh和en，分别表示中文和英文。为了覆盖这些语言大类下的各种分支，通过添加各种子标签，我们可以更具体地描述某种语言。比如，添加语言的脚本（script）标签，zh-Hans和zh-Hant分别表示简体中文和繁体中文；通过添加国家/地区标签，en-US,en-GB分别表示美国英语和英国英语；类似的，zh-CN、zh-TW和zh-HK则分别表示中国大陆简体中文，中国台湾繁体中文和中国香港繁体中文。
区域代码（locale code）则用来表示某一个区域的语法形式和语言代码类似。比如zh、en等。另外一种常见的形式是附加了国家/地区代码的形式，其使用下划线将语言代码与国家/区域连接起来，比如en_US,zh_CN、zh_TW，另外还有添加语言脚本的形式，比如 zh_hans_cn、zh_hant_TW等

语言代码中添加国家/区域代码时的连接符是连接线，而区域代码使用下划线。

文本的国际化

flash(u"文章发表成功")

# 替换成
from flask_babel import gettext
flash(gettext(u"文章发表成功"))

# 或者使用 _ 代替gettext
from flask_babel import _
flash(_(u"文章发表成功"))

对于英文等单复数包含单词变化的语言，使用ngettext

传入两个字符串，依次为单数形式和复数形式的字符串，其中包含用于区分单复数的变量num，第三个参数则传入代表num数值的变量。

1	flash(ngettext(u"%(num)s Apple" , u"%(num)s Apples" , num=number_of_apples))

注意 : _只会在请求上下文中执行 , 如果想在请求上下文外执行 , 使用lazy_gettext
1
2
3
4
from flask_babel import Babel, lazy_gettext as _l

class LoginForm(FlaskForm):
    username = StringField(_l("Username"))

在模板中使用

<h1>Join now</h1> 

{# 替换成 #}
<h1>{{ _("Join now") }}</h1> 
<h1>{{ _("Welcome , %(username)s", username=current_user.username) }}</h1>

设置子域名

使用subdomain参数

api_v1 = Blueprint('api_v1', __name__)

app.register_blueprint(api_v1, url_prefix='/api/v1')
app.register_blueprint(api_v1, url_prefix='/v1', subdomain='api')

用于注册路由的 routed装饰器也接收 subdomain参数，可以为某个视图定义子域。

flask-cors 支持跨域资源共享

同源策略（Same origin policy）: 出于安全考虑，浏览器会限制从脚本内发起的跨域请求。这里的跨域包括不同域名、不同端口、不同HTTP模式（HTTP、Https等）。

在CORS流行之前，大多数AP都通过支持 JSONP（json with Padding）来支持跨域请求。和 JSONP相比，CORS更加方便灵活，支持更多的跨域请求方法，并且在2014年成为W3C的推荐标准，逐渐开始替代 JSONP。

CORS 原理简介: https://www.ruanyifeng.com/blog/2016/04/cors.html

CORS需要浏览器和服务器同时支持。

整个CORS通信过程，都是浏览器自动完成，不需要用户参与。对于开发者来说，CORS通信与同源的AJAX通信没有差别，代码完全一样。浏览器一旦发现AJAX请求跨源，就会自动添加一些附加的头信息，有时还会多出一次附加的请求，但用户不会有感觉。

因此，实现CORS通信的关键是服务器。只要服务器实现了CORS接口，就可以跨源通信。

1	pip install flask-cors

简单使用:

直接注册相关蓝图即可.

from flask import Blueprint
from flask_cors import CORS

api_v1 = Blueprint("api_v1",__name__)
CORS(api_v1)

默认情况下，Flask-CORS会为蓝本下的所有路由添加跨域请求支持，并且允许来自任意源的跨域请求。

RESTfull API

Resource Representational State Transfer。
这可以理解为“资源（Resource）在网络中以某种表现形式（Representation）进行状态转移（State Transfer）”。

资源端点与HTTP方法的操作含义

api.example.com/users：所有用户。
api.example.com/users/123/：id为123的用户。
api.example.com/users//123/posts：id为33的用户的所有文章。
api.example.com/posts：所有文章。
api.example.com/posts/23：id为23的文章。
api.example.com/posts/23/comments：id为23的文章的所有评论。

URL	GET	PUT	PATCH	POST	DELETE
资源集合比如http://api.example.com/posts	列出集合成员的所有信息	替换整个集合的资源	一般不使用	在集合中创建个新条目，新条目的URL自动生成并包含在响应中返回	删除整个资源
单个元素比如http://api.example.com/posts/123	获取指定资源的详细信息，采用XML或JSON等表现形式	替换指定的集合成员，如果不存在则创建	更新集合成员，仅提供更新的内容	一般不使用	删除指定的集合成员

MethodView

类似于Django的Class View

from flask.views import MethodView

api_v1 = Blueprint('api_v1', __name__)

def auth_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token_type, token = get_token()
        if request.method != 'OPTIONS':
            if token_type is None or token_type.lower() != 'bearer':
                return api_abort(400, 'The token type must be bearer.')
            if token is None:
                return token_missing()
            if not validate_token(token):
                return invalid_token()
        return f(*args, **kwargs)
    return decorated

class ItemAPI(MethodView):
    decorators = [auth_required]

    def get(self, item_id):
        """Get item."""
        item = Item.query.get_or_404(item_id)
        if g.current_user != item.author:
            return api_abort(403)
        return jsonify(item_schema(item))

    def put(self, item_id):
        """Edit item."""
        item = Item.query.get_or_404(item_id)
        if g.current_user != item.author:
            return api_abort(403)
        item.body = get_item_body()
        db.session.commit()
        return '', 204

    def patch(self, item_id):
        """Toggle item."""
        item = Item.query.get_or_404(item_id)
        if g.current_user != item.author:
            return api_abort(403)
        item.done = not item.done
        db.session.commit()
        return '', 204

    def delete(self, item_id):
        """Delete item."""
        item = Item.query.get_or_404(item_id)
        if g.current_user != item.author:
            return api_abort(403)
        db.session.delete(item)
        db.session.commit()
        return '', 204
    
api_v1.add_url_rule('/user/items', view_func=ItemsAPI.as_view('items'), methods=['GET', 'POST'])

API的 OAuth 认证及第三方认证登录

用户通过一次认证后，在服务器端为用户生成一个认证令牌，在之后的请求中，客户端可以通过认证令牌进行认证。出于安全的考虑，认证令牌还会设置过期时间。

from itsdangerous import TimedJSONWebSignatureSerializer as Serializer, BadSignature, SignatureExpired

def generate_token(user):
    expiration = 3600
    s = Serializer(current_app.config['SECRET_KEY'], expires_in=expiration)
    token = s.dumps({'id': user.id}).decode('ascii')
    return token, expiration

class AuthTokenAPI(MethodView):
    def post(self):
        grant_type = request.form.get('grant_type')
        username = request.form.get('username')
        password = request.form.get('password')

        if grant_type is None or grant_type.lower() != 'password':
            return api_abort(code=400, message='The grant type must be password.')
        user = User.query.filter_by(username=username).first()
        if user is None or not user.validate_password(password):
            return api_abort(code=400, message='Either the username or password was invalid.')
        token, expiration = generate_token(user)
        response = jsonify({
            'access_token': token,
            'token_type': 'Bearer',
            'expires_in': expiration
        })
        response.headers['Cache-Control'] = 'no-store'
        response.headers['Pragma'] = 'no-cache'
        return response

请求参数一般有4个:

grant_ type
username
password
scope : 允许的权限范围（可选）

返回的token_type : access令牌是客户端用于访问受登录保护时的认证凭证，令牌类型中的“Bearer”是RFC6750定义的令牌类型（常被翻译为不记名令牌）。

之后我们就可以使用View的装饰器了:

def get_token():
    if 'Authorization' in request.headers:
        try:
            token_type, token = request.headers['Authorization'].split(None, 1)
        except ValueError:
            token_type = token = None
    else:
        token_type = token = None
    return token_type, token

def api_abort(code, message=None, **kwargs):
    if message is None:
        message = HTTP_STATUS_CODES.get(code, '')
    response = jsonify(code=code, message=message, **kwargs)
    response.status_code = code
    return response

def invalid_token():
    response = api_abort(401, error='invalid_token', error_description='Either the token was expired or invalid.')
    response.headers['WWW-Authenticate'] = 'Bearer'
    return response

def token_missing():
    response = api_abort(401)
    response.headers['WWW-Authenticate'] = 'Bearer'
    return response

def auth_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token_type, token = get_token()
        # Flask通常单独处理OPTIONS请求，但在配置为将这些请求转发给应用程序的情况下，我们需要忽略身份验证头并让请求通过，以避免与CORS发生不必要的交互。
        if request.method != 'OPTIONS':
            if token_type is None or token_type.lower() != 'bearer':
                return api_abort(400, 'The token type must be bearer.')
            if token is None:
                return token_missing()
            if not validate_token(token):
                return invalid_token()
        return f(*args, **kwargs)
    return decorated

发送请求时需要把认证令牌附加在请求首部的 Authorization字段中，并且在令牌前指定令牌类型（即Bearer）

Web API的序列化与反序列化

对于Web API来说，

序列化（Serialize）就是把数据库模型对象转换成JSON数据。
反序列化（Deserialize）就是把JSON数据转换成数据库模型对象。

JSON数据风格指南摘要

https://github.com/darcyliu/google-styleguide/blob/master/JSONStyleGuide.md

数组类型应该是复数属性名。其它属性名都应该是单数。
日期应该使用RFC3339建议的格式 , 时间间隔应该使用ISO 8601建议的格式
1
2
3
4
{
"lastUpdate": "2007-11-06T16:34:41.000Z",
"duration": "P3Y6M4DT12H30M5S"
}

Marshmallow 与 Webargs

Marshmallow的用法和 WTForms类似，通过创建模式类事先定义好字段类型以及验证函数。
对模式类对象调用dump()方法和 load方法分别执行序列化和反序列化操作，同时对传入的对象进行验证，并返回验证后的数据字典和相应的错误消息字典。
Webargs可以解析请求中包含的表单、查询字符串、JSON、cookies、files、首部字段等系列数据，然后根据预定的样式进行验证，如果验证未通过会生成内置的错误消息。

webargs文档 : https://webargs.readthedocs.io/en/latest/

默认情况下，webargs将以JSON形式从请求正文中搜索参数。
你也可以在 use_args裝饰器中使用 location参数显式指定一个数据位置，可用的值为 :
- querystring（等同于query，表示查询字符串）、
- json（表示JSON）、
- form（表示表单数据）、
- headers（表示首部字段）、
- cookies（表示Cookie）
- files（表示文件）。

webargs的422错误

当验证出错时，Webargs会返回422响应（Unprocessable Entity，表示实体无法处理，即语义错误），我们可以注册一个对应的错误处理函数，如下所示：

@app.errorhandler(422)
def handle_validation_error(e):
    exc = e.exc
    return jsonify({'errors':exc.messages}) , 422

webargs解析字段一览

from webargs import fields, validate

user_args = {
    # Required arguments
    "username": fields.Str(required=True),
    # Validation
    "password": fields.Str(validate=lambda p: len(p) >= 6),
    # OR use marshmallow's built-in validators
    "password": fields.Str(validate=validate.Length(min=6)),
    # Default value when argument is missing
    "display_per_page": fields.Int(missing=10),
    # Repeated parameter, e.g. "/?nickname=Fred&nickname=Freddie"(重复字段)
    "nickname": fields.List(fields.Str()),
    # Delimited list, e.g. "/?languages=python,javascript"(带括号字段)
    "languages": fields.DelimitedList(fields.Str()),
    # When value is keyed on a variable-unsafe name
    # or you want to rename a key(重命名字段)
    "user_type": fields.Str(load_from="user-type"),
    # OR, on marshmallow 3
    # "user_type": fields.Str(data_key="user-type"),
    # 嵌套字段
    "name": fields.Nested(
        {"first": fields.Str(required=True), "last": fields.Str(required=True)}
    )
}

webargs配合marshmallow

from marshmallow import Schema, fields
from webargs.flaskparser import use_args

class UserSchema(Schema):
    id = fields.Int(dump_only=True)  # read-only (won't be parsed by webargs)
    username = fields.Str(required=True)
    password = fields.Str(load_only=True)  # write-only
    first_name = fields.Str(missing="")
    last_name = fields.Str(missing="")
    date_registered = fields.DateTime(dump_only=True)

@use_args(UserSchema())
def profile_view(args):
    username = args["userame"]
    # ...

@use_kwargs(UserSchema())
def profile_update(username, password, first_name, last_name):
    update_profile(username, password, first_name, last_name)
    # ...

webargs多字段位置

# webargs 5.x and older
@parser.use_args(
    {
        "q1": ma.fields.Int(location="query"),
        "q2": ma.fields.Int(location="query"),
        "h1": ma.fields.Int(location="headers"),
    },
    locations=("query", "headers"),
)
def foo(q1, q2, h1):
    ...


# webargs 6.x
@parser.use_args({"q1": ma.fields.Int(), "q2": ma.fields.Int()}, location="query")
@parser.use_args({"h1": ma.fields.Int()}, location="headers")
def foo(q1, q2, h1):
    ...

第十一章在线聊天室

使用 Flask-SocketIO 建立实时双向通信

Socket.IO（http://socket.io）是一个基于 Web Socket实现实时通信的开源 JavaScript库，它可以简化实时web程序（real-time application,RTA）的开发过程。
它除了支持 Web Socket之外，还支持许多种轮询（Polling）机制以及其他模拟实时通信的方式。
SocketIO会根据浏览器对通信机制的支持情况自动选择最佳的方式来实现实时通信，实现了对浏览器的“降级支持”。
SocketIO为这些通信方式实现了统一的接口，因此使用起来非常简单。除此之外，SocketIO还提供了诸如广播、命名空间、房间、存储客户端数据、异步IO等非常方便的功能。

1 2	pip install flask-socketio pip install eventlet

from flask_socketio import SocketIO

socketio = SockerIO()
socketio.init_app(app)

启动Socket.IO服务器

为了正确启动 Socket.IO服务器，Flask-Socket覆写了 Flask提供的 flask run命令，我们可以像往常一样使用 flask run命令启动 SocketIO服务器

另外，我们也可以直接调用 socketio对象提供的run方法来启动服务器，传入程序实例app作为参数：

from flask import Flask
from flask_sockerio import SocketIO

app = Flask(__name__)
app.config['SECRET_KEY'] = 'serect string'
socketio = SocketIO(app)

if __name__ == '__main__':
    socketio.run(app)

在客户端，我们需要加载 Socket.IO资源并调用io()方法与我们在服务器端运行的 SocketIO服务器建立连接。首先我们需要加载对应的资源文件，你可以访问https://cdnjs.com/libraries/socket.io将资源文件下载到本地 , 之后就可以使用io()了
1
2
3
4
5
6
<script type="text/javascript">
var socket = io();
// 如果不传入任何参数，会默认使用服务器的根URL，
// 等价于下面
// var socket = io('/');
</script>

SocketIO事件

消息流程

用户发送消息请求到服务器端。
服务器端接收消息请求并把消息广播给所有客户端。
所有客户端接收消息并显示。

在 SocketIO中，服务器端和客户端之间交流的数据被称为 SocketIO事件（event），这里的事件就是包含特定信息的数据，类似我们常说的请求/响应
简单来说，在 Socket.IO中，双向通信是这样实现的：
1. 客户端通过调用emit函数来将一个事件发送到服务器端，并传入数据作为参数，这会触发服务器端创建对应的事件处理函数。
2. 而服务器端也可以通过调用 emit函数向客户端发送事件，并传人数据作为参数，类似地，这会触发客户端创建对应的事件处理函数。

简单示例

<textarea rows="2" id="message-textarea" placeholder="Write your message here... Enter to send"></textarea>

<script src="{{ url_for('static', filename='js/jquery.min.js') }}"></script>
<script src="{{ url_for('static', filename='js/socket.io.min.js') }}"></script>
<script type="text/javascript">
    var socket = io();
</script>

<script>
    var ENTER_KEY = 13 ;
    function new_message(e) {
        var $textarea = $('#message-textarea');
        var message_body = $textarea.val().trim();
        if (e.which === ENTER_KEY && !e.shiftKey && message_body) {
            e.preventDefault();
            socket.emit('new message', message_body);
            $textarea.val('')
        }
    }
    $('#message-textarea').on('keydown', new_message.bind(this));
</script>

<script>
    // 接收
    socket.on('new message', function (data) {
        $('.messages').append(data.message_html);
    });
</script>

Fask-SocketIo提供on()装饰器来注册用于接收客户端发来事件的事件处理函数。
创建new message事件处理函数，用来处理客户端发送的new message事件

from catchat.extensions import socketio, db

@socketio.on('new message')
def new_message(message_body):
    sid = request.sid  # io客户端的sid, socketio用此唯一标识客户端.
    html_message = to_html(message_body)
    message = Message(author=current_user._get_current_object(), body=html_message)
    db.session.add(message)
    db.session.commit()
    emit('new message',
         {'message_html': render_template('chat/_message.html', message=message),
          'message_body': html_message,
          'gravatar': current_user.gravatar,
          'nickname': current_user.nickname,
          'user_id': current_user.id},
         broadcast=True)

在emit()函数中，
1. 第一个参数用来指定事件名称，当服务器端的 emit函数被调用时，会触发已连接客户端中对应的事件处理函数。
2. 第二个参数是我们要发送的数据，要发送的数据根据客户端的需要而定，这里我们使用 render
  template函数渲染存储单个消息HTML代码的message。html模板，在客户端使用 JavaScript可以直接将这个数据插入到消息列表的HTML元素中。事件中包含的数据类型可以为字符串、列表或字典。当数据的类型为列表或字典时，会被序列化为JSON格式。
3. 因为我们要把消息发送给所有已经连接的客户端，所以需要将broadcast参数设为True，这样就会广播这个事件，即将事件发送给所有已连接的客户端。

储存global数据

如果不存入redis的话 , 使用global关键字

@socketio.on('connect')
def connect():
    global online_users
    if current_user.is_authenticated and current_user.id not in online_users:
        online_users.append(current_user.id)
    emit('user count', {'count': len(online_users)}, broadcast=True)

@socketio.on('disconnect')
def disconnect():
    global online_users
    if current_user.is_authenticated and current_user.id in online_users:
        online_users.remove(current_user.id)
    emit('user count', {'count': len(online_users)}, broadcast=True)

1
2
3

socket.on('user count', function (data) {
    $('#user-count').html(data.count);
});

connect连接事件处理程序可以选择返回False以拒绝连接。这样就可以在此时对客户端进行身份验证。

异常处理

@socketio.on_error()        # Handles the default namespace
def error_handler(e):
    pass

@socketio.on_error('/chat') # handles the '/chat' namespace
def error_handler_chat(e):
    pass

@socketio.on_error_default  # handles all namespaces without an explicit error handler
def default_error_handler(e):
    pass

gunicorn配置

通过gunicorn启动eventlet服务器的命令行

1	gunicorn --worker-class eventlet -w 1 module:app

使用gevent，

1	gunicorn -k gevent -w 1 module:app

Socket.io通信频道分离

命名空间

简单地说，不同的命名空间就是不同的URL路径，比如/foo和/bar就是两个不同的命名空间。
可以把 SocketIo中的命名空间比作 Flask中的蓝本
可以通过request.namesapce获取

1
2
3

@socketio.on('my event',namespace='/chat')
def handle_my_custom_event(json):
    emit('my response', json, namespace='/chat')

1	var socket = io.connect('/anoymous')

房间

对于许多应用程序，有必要将用户分组为可以一起寻址的子集。
最好的例子是具有多个房间的聊天应用程序，其中用户从他们所在的房间或房间接收消息，而不是从其他用户所在的其他房间接收消息。
SocketIO支持通过房间的概念join_room()和leave_room()功能：这两个函数分别用来把当前用户（客户端）加入和退出房间。你还可以使用 close room函数来删除个房间，并清空其中的用户。另外，rooms函数可以返回某个房间内的客户端列表。

from flask_socketio import join_room, leave_room

@socketio.on('join')
def on_join(data):
    username = data['username']
    room = data['room']
    ## 传入房间的唯一标识
    join_room(room)
    send(username + ' has entered the room.', room=room)

@socketio.on('leave')
def on_leave(data):
    username = data['username']
    room = data['room']
    leave_room(room)
    send(username + ' has left the room.', room=room)
    
@socketio.on('room message')
def new_room_message(message_body):
    username = data['username']
    room = data['room']
    ## 传入房间的唯一标识
    join_room(room)
    emit('message', {'message':current_user.username+":"+message_body}, room='current_user.room')

基于房间你也可以实现私信私聊功能。只需要把room设为代表某个用户的唯一值，在发送事件时，就只有目标用户的客户端才能接收到事件。你可以把这种实现方法理解为“一个人的房间”。这个能代表用户的唯一值可以是主键值、username或是 Flask-SocketIo附加到 request对象上代表每个客户端id的 session id（request.sid）。

使用 Flask-OAuthlib 实现第三方登录

第三方登录的实质是借助第三方服务开放的 Web APl进行 OAuth授权。授权成功后，我们就可以通过第三方服务的 Web api获取到用户的资料以及其他资源。

1	pip install flask-oauthlib

在 OAuth认证中 :

需要获取和使用资源的一方，即我们的程序，通常被称为资源消费者（Resource Consumer）或客户端（Client）。
拥有用户资源的各种在线服务提供方则被称为资源提供者（Resource provider）。
用户被称为资源拥有者（Resource Owner）。

简单使用

流程:

用户点击登录 , 调用oauth_login视图 , 该视图重定向到第三方登录页面
用户键入 , 点击登录 , 第三方应用会将用户重定向到我们注册 OAuth程序时提供的回调URL(例如http://127.0.0.1/callback/github) , 并且返回access
视图接收到这个回调请求，获取code，发送一个POST请求到用于获取access 令牌的URL，进行验证 .
验证成功则说明登录成功

在 OAuth认证中，

我们开发的程序也被称为本地程序（local application），
我们要与之交互的第三方服务提供方则相应被称为远程程序（remote application）。

<div class="ui message">
    <a class="icon" href="{{ url_for('oauth.oauth_login', provider_name='github') }}">
        <i class="github big icon black"></i>
    </a>
    <a class="icon" href="{{ url_for('oauth.oauth_login', provider_name='google') }}">
        <i class="google big icon red"></i>
    </a>
    <a class="icon" href="{{ url_for('oauth.oauth_login', provider_name='twitter') }}">
        <i class="twitter big icon blue"></i>
    </a>
</div>

from catchat.extensions import oauth

github = oauth.remote_app(
    # 在这些参数中，不同服务提供方的各种URL可以在各自的开发文档中看到。
    name='github',
    consumer_key=os.getenv('GITHUB_CLIENT_ID'),
    consumer_secret=os.getenv('GITHUB_CLIENT_SECRET'),
    request_token_params={'scope': 'user'},
    base_url='https://api.github.com/',
    request_token_url=None,
    access_token_method='POST',
    access_token_url='https://github.com/login/oauth/access_token',
    authorize_url='https://github.com/login/oauth/authorize',
)
twtter = oauth.remote_app(...)
google = oauth.remote_app(...)
providers = {
    'github': github,
    'google': google,
    'twitter': twitter
}

@oauth_bp.route('/login/<provider_name>')
def oauth_login(provider_name):
    if provider_name not in providers.keys():
        abort(404)
    if current_user.is_authenticated:
        return redirect(url_for('chat.home'))
    callback = url_for('.oauth_callback', provider_name=provider_name, _external=True)
    return providers[provider_name].authorize(callback=callback)


@oauth_bp.route('/callback/<provider_name>')
def oauth_callback(provider_name):
    if provider_name not in providers.keys():
        abort(404)
    provider = providers[provider_name]
    # 将access的数据POST到第三方应用进行验证 , 得到返回响应
    response = provider.authorized_response()
    if response is not None:
        if provider_name == 'twitter':
            access_token = response.get('oauth_token'), response.get('oauth_token_secret')
        else:
            access_token = response.get('access_token')
    else:
        access_token = None
    if access_token is None:
        flash('Access denied, please try again.')
        return redirect(url_for('auth.login'))
    username, website, github, email, bio = get_social_profile(provider, access_token)
    user = User.query.filter_by(email=email).first()
    if user is None:
        user = User(email=email, nickname=username, website=website,
                    github=github, bio=bio)
        db.session.add(user)
        db.session.commit()
        login_user(user, remember=True)
        return redirect(url_for('chat.profile'))
    login_user(user, remember=True)
    return redirect(url_for('chat.home'))

# API的资源端点
profile_endpoints = {
    'github': 'user',
    'google': 'userinfo',
    'twitter': 'account/verify_credentials.json?include_email=true'
}

def get_social_profile(provider, access_token):
    profile_endpoint = profile_endpoints[provider.name]
    # 对远程程序对象调用get方法来获取用户的部分信息
    # 第一个参数url是发送请求的URL，即API的资源端点。
    # 在github中 ,传入的资源URL为user，结合我们注册 github远程程序时传入的baseurl，最终生成的端点URL即http://api.github.com/user，这个端点用来获取GitHub用户的信息
    response = provider.get(profile_endpoint, token=access_token)
    if provider.name == 'twitter':
        username = response.data.get('name')
        website = response.data.get('url')
        github = ''
        email = response.data.get('email')
        bio = response.data.get('description')
    elif provider.name == 'google':
        username = response.data.get('name')
        website = response.data.get('link')
        github = ''
        email = response.data.get('email')
        bio = ''
    else:
        username = response.data.get('name')
        website = response.data.get('blog')
        github = response.data.get('html_url')
        email = response.data.get('email')
        bio = response.data.get('bio')
    return username, website, github, email, bio

我们首先需要创建用于第三方登录的视图，在视图函数中使用redirect重定向到服务提供方的授权URL，并附加相应的查询参数。但我们不必手动做这个工作，因为 Flask-OAuthlib为每一个注册后的远程程序对象提供了 authorized方法，这个方法用来构建授权URL并生成重定向响应。

github.authorize(url_for('.github_callback',_external=True))
response = provider.authorized_response()封装了POST验证access的一系列操作

bleach+markdown库支持markdown

markdown : 文本转换
Bleach : HTML清理工具包 (防止恶意内容 , 比如 JavaScript脚本。)

1 2	pip install markdown pip install bleach

>>> import markdown
>>> md = "# hello markdown"
>>> markdown.markdown(md)
'<h1>hello markdown</h1>'

简单使用

Bleach的清理工作是基于白名单 (allowed_tags , allowed_attributes)进行的。

from bleach import clean, linkify
from markdown import markdown

def to_html(raw):
    allowed_tags = ['a', 'abbr', 'b', 'br', 'blockquote', 'code',
                    'del', 'div', 'em', 'img', 'p', 'pre', 'strong',
                    'span', 'ul', 'li', 'ol']
    allowed_attributes = ['src', 'title', 'alt', 'href', 'class']
    html = markdown(raw, output_format='html',
                    # 使用fenced_code和codehilite插件进行代码语法高亮
                    extensions=['markdown.extensions.fenced_code',
                                'markdown.extensions.codehilite'])
    clean_html = clean(html, tags=allowed_tags, attributes=allowed_attributes)
    return linkify(clean_html)

接收用户输入的包含 Markdown标记的源文本。
将 Markdown文本转换为HTML格式。
将转换好的HTML文本渲染到模板中。

代码语法高亮

Flask-CKEditor内置了代码语法高亮功能 , 这里我们使用pyments

1	pip install pyments

原理 : 对于HTML格式来说，通过解析代码片段的语法结构，Payments会使用标签分隔每一个语法单元，并添加对应的样式类，最后通过加载对应的CSS文件即可实现代码“上色”。

desktop notification(桌面通知)

document.addEventListener('DOMContentLoaded', function () {
    if (!Notification) {
        alert('Desktop notifications not available in your browser.');
        return;
    }
    if (Notification.permission !== "granted")
        Notification.requestPermission();
});

Notification.permission存储用户的许可状态值 : granted表示允许，denied表示拒绝，默认为 default（等同于 denied）。
如果 Notification. permission的值不是 granted,那么就调用 Notification.requestPermission()方法请求授权,这会在用户浏览器中弹出一个授权请求窗口.

简单使用

notification.onclick属性定义的函数会在提醒弹窗被单击时执行
notification.close()方法关闭弹窗。

const NotificationInstance = Notification || window.Notification;
if (!!NotificationInstance) {
    const permissionNow = NotificationInstance.permission;
    if (permissionNow === 'granted') {//允许通知
        CreatNotification();
    } else if (permissionNow === 'denied') {
        console.log('用户拒绝了你!!!');
    } else {
        setPermission();
    }
    function setPermission() {
        //请求获取通知权限
        NotificationInstance.requestPermission(function (PERMISSION) {
            if (PERMISSION === 'granted') {
                CreatNotification();
            } else {
                console.log('用户无情残忍的拒绝了你!!!');
            }
        });
    }
    function CreatNotification() {
        const n = new NotificationInstance('XX网站消息通知', {
            body: '你的朋友有新状态啦，快去围观吧！',
            tag: '2ue',
            icon: 'https://2ue.github.io/images/common/avatar.png',
            data: {
                url: 'https://2ue.github.io'
            }
        });
        n.onshow = function () {
            console.log('通知显示了！');
        };
        n.onclick = function (e) {
            //可以直接通过实例的方式获取data内自定义的数据
            //也可以通过访问回调参数e来获取data的数据
            window.open(n.data.url, '_blank');
            n.close();
        };
        n.onclose = function () {
            console.log('你墙壁了我！！！');
        };
        n.onerror = function (err) {
            console.log('出错了，小伙子在检查一下吧');
            throw err;
        };
        setTimeout(() => {
            n.close();
        }, 2000);
    }
}

第七章 留言板

使用click创建虚拟数据

Bootstrap-Flask 插件

Flask-Moment 本地化日期和时间

第八章 个人博客

为什么使用Endpoint?

邻接列表关系

渲染导航链接

切换博客主题(切换CSS)

Flask-Login

login_user的背后逻辑

currenr_user

为一整个蓝图添加login_required

使用CSRFProtect拓展手动实现CSRF防御

取消对某个蓝图的CSRF防御

第九章 图片社交网站

项目组织架构

在线的占位图片服务

WTForm的EqualTo和Regexp校验器

生成JWS（json Web Signature）

flask的装饰器必须使用wraps装饰器

flash函数中返回危险文本的操作

基于用户角色的权限管理 RBAC (Role-Based Access Control)

角色和权限model设计

替换AnonymousUserMixin

权限装饰器 及 装饰器的继承(装饰器里调用装饰器)

Flask-Dropzone 优化文件上传

相关宏

重要配置

简单示例

Flask-Avatars 处理用户头像

相关宏

文件删除

当页编辑的最佳实践

注册全局Ajax错误

flask-login的fresh_login_required : 重新登录

subquery 与 in_ 与 join

sqlalchemy随机获取内容

Flask Whooshee 实现全文搜索

示例

第十章 代办事项程序

单页面应用布局

在根页面内切换子页面

template的True,False转为true,false

使用 Flask-Babel 支持国际化与本地化

区域和语言

文本的国际化

设置子域名

flask-cors 支持跨域资源共享

RESTfull API

资源端点与HTTP方法的操作含义

MethodView

API的 OAuth 认证 及第三方认证登录

Web API的序列化与反序列化

JSON数据风格指南摘要

Marshmallow 与 Webargs

webargs的422错误

webargs解析字段一览

webargs配合marshmallow

webargs多字段位置

第十一章 在线聊天室

使用 Flask-SocketIO 建立实时双向通信

启动Socket.IO服务器

SocketIO事件

简单示例

储存global数据

异常处理

gunicorn配置

Socket.io通信频道分离

命名空间

房间

使用 Flask-OAuthlib 实现第三方登录

简单使用

bleach+markdown库支持markdown

简单使用

代码语法高亮

desktop notification(桌面通知)

简单使用

第七章留言板

第八章个人博客

第九章图片社交网站

权限装饰器及装饰器的继承(装饰器里调用装饰器)

第十章代办事项程序

API的 OAuth 认证及第三方认证登录

第十一章在线聊天室