FlaskWeb开发（2）Flask实例

《FlaskWeb开发》阅读笔记

第二部分 Flask实例：社交博客 —— 用户认证、角色分配与资料编辑

第二部分实例：社交博客(上)

用户认证

认证包：

Flask-Login：管理已登录用户的用户会话
Werkzeug：计算密码散列值并进行核对
itsdangerous：生成并核对加密安全令牌

密码散列实现（Werkzeug）

generate_password_hash(password, method=pbkdf2:sha1, salt_length=8)：原始密码作为输入，以字符串形式输出密码的散列值
check_password_hash(hash, password)：参数为密码散列和用户输入的密码

from werkzeug.security import generate_password_hash, check_password_hash 
 
class User(db.Model): 
    # ... 
    password_hash = db.Column(db.String(128)) 
 
    @property  # 让password字段无法直接读取
    def password(self):  
        raise AttributeError('password is not a readable attribute') 

    @password.setter  # 修改用户的password 字段 
    def password(self, password):  # 通过password的只写属性实现
        self.password_hash = generate_password_hash(password)
    def verify_password(self, password): 
        return check_password_hash(self.password_hash, password)

$ tree /F
│  .gitignore
│  config.py
│  flasky.py
│  LICENSE
│  README.md
│  requirements.txt
├─app
│  │  email.py
│  │  models.py
│  │  __init__.py
│  │
│  ├─main
│  │      errors.py
│  │      forms.py
│  │      views.py
│  │      __init__.py
│  │
│  ├─static
│  │      favicon.ico
│  │
│  └─templates
│      │  404.html
│      │  500.html
│      │  base.html
│      │  index.html
│      │
│      └─mail
│              new_user.html
│              new_user.txt
│
├─migrations
│  │  alembic.ini
│  │  env.py
│  │  README
│  │  script.py.mako
│  │
│  └─versions
│          38c4e85512a9_initial_migration.py
│
└─tests
        test_basics.py
        test_user_model.py
        __init__.py

认证蓝本

auth 蓝本保存在同名 Python 包

蓝本的包构造文件创建蓝本对象，再从 views.py 引入路由

# app/auth/__init__.py：创建蓝本
from flask import Blueprint 
 
auth = Blueprint('auth', __name__) 
 
from . import views

使用蓝本的 route 修饰器定义与认证相关的路由

from flask import render_template 
from . import auth 
 
@auth.route('/login') 
def login(): 
    return render_template('auth/login.html')  # 模板的路径是相对于程序模板文件夹，这里保存在了一个单独的文件夹

蓝本要在create_app()中附加到程序

# app/__init__.py：附加蓝本
def create_app(config_name): 
    # ... 
    from .auth import auth as auth_blueprint 
    app.register_blueprint(auth_blueprint, url_prefix='/auth')  # 有了url_prefix，注册后蓝本中定义的所有路由都会加上指定的前缀，即http://localhost:5000/auth/login
 
    return app

Flask-Login认证

需要记录用户认证状态
User 模型必须实现：
- is_authenticated()：若用户已登录，返回 True，否则返回 False
- is_active()：允许用户登录，返回 True，否则返回 False；要禁用账户，可以返回 False
- is_anonymous()：普通用户返回 False
- get_id()：返回用户的唯一标识符——使用 Unicode 编码的字符串

Flask-Login 提供 UserMixin 类，包含以上的实现

from flask.ext.login import UserMixin 
 
class User(UserMixin, db.Model): 
    __tablename__ = 'users' 
    id = db.Column(db.Integer, primary_key = True) 
    email = db.Column(db.String(64), unique=True, index=True)  # 用户使用email登录
    username = db.Column(db.String(64), unique=True, index=True) 
    password_hash = db.Column(db.String(128)) 
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

在create_app()中实现

from flask.ext.login import LoginManager 
 
login_manager = LoginManager() 
login_manager.session_protection = 'strong'  # 属性可以设为 None、'basic' 或 'strong'，提供不同的安全等级
login_manager.login_view = 'auth.login'  # 设置登录页面，auth为蓝本名字
 
def create_app(config_name): 
    # ... 
    login_manager.init_app(app) 
    # ...

要求实现回调函数，接收 user_id 加载用户

# app/models.py：加载用户的回调函数
from . import login_manager 
 
@login_manager.user_loader 
def load_user(user_id): 
    return User.query.get(int(user_id))  # 如果能找到用户，则返回用户对象

只让认证用户访问路由：

from flask.ext.login import login_required 
 
@app.route('/secret') 
@login_required  # 使用此修饰器
def 。。。

设置登录表单

设置登入和登出的视图函数

# app/auth/views.py
from flask import render_template, redirect, request, url_for, flash 
from flask.ext.login import login_user, login_required 
from . import auth 
from ..models import User 
from .forms import LoginForm 
 
@auth.route('/login', methods=['GET', 'POST']) 
def login(): 
    form = LoginForm() 
    if form.validate_on_submit(): 
        user = User.query.filter_by(email=form.email.data).first()  # 使用表单中填写的email从数据库中加载用户
        if user is not None and user.verify_password(form.password.data):  # 校验密码
            login_user(user, form.remember_me.data)  # 用户会话中把用户标记为已登录
            return redirect(request.args.get('next') or url_for('main.index'))  # 用户访问未授权的URL时会显示登录表单，此时将原地址保存在查询字符串的 next 参数中。如果不是上面的情况，则重定向到首页
        flash('Invalid username or password.')  # 用户输入的电子邮件或密码不正确时，显示此flash消息
    return render_template('auth/login.html', form=form)


 
@auth.route('/logout') 
@login_required 
def logout(): 
    logout_user()  # 删除并重设用户会话
    flash('You have been logged out.')  # 随后显示一个Flash消息，以确认这次操作
    return redirect(url_for('main.index'))

注册新用户

添加注册表单

表单函数

# app/auth/forms.py：用户注册表单
from flask.ext.wtf import Form
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import Required, Length, Email, Regexp, EqualTo
from wtforms import ValidationError
from ..models import User
class RegistrationForm(Form): 
    email = StringField('Email', validators=[Required(), Length(1, 64), 
                                             Email()]) 
    username = StringField('Username', validators=[ 
        Required(), Length(1, 64), Regexp('^[A-Za-z][A-Za-z0-9_.]*$', 0, 
                                          'Usernames must have only letters, ' 
                                          'numbers, dots or underscores')])  # Regexp 验证函数，确保username字段只包含字母、数字、下划线和点号；正则表达式后的参数：正则表达式的旗标和验证失败时显示的错误消息
    password = PasswordField('Password', validators=[ 
        Required(), EqualTo('password2', message='Passwords must match.')])  # Equalto：验证两个密码字段中的值是否一致
    password2 = PasswordField('Confirm password', validators=[Required()]) 
    submit = SubmitField('Register') 

  # 表单类中定义validate_ 开头且后面跟着字段名的方法——类似验证函数，和验证函数一起调用
    def validate_email(self, field): 
        if User.query.filter_by(email=field.data).first(): 
            raise ValidationError('Email already registered.')  # 验证失败则抛出ValidationError异常，参数是错误消息

    def validate_username(self, field): 
        if User.query.filter_by(username=field.data).first(): 
            raise ValidationError('Username already in use.')

登录页面要显示一个指向注册页面的链接

<!--app/templates/auth/login.html：链接到注册页面-->
<p> 
    New user? 
    <a href="{{ url_for('auth.register') }}"> 
        Click here to register 
    </a> 
</p>

注册的视图函数

# app/auth/views.py：用户注册路由
@auth.route('/register', methods=['GET', 'POST']) 
def register(): 
    form = RegistrationForm() 
    if form.validate_on_submit(): 
        user = User(email=form.email.data, 
                    username=form.username.data, 
                    password=form.password.data) 
        db.session.add(user) 
        flash('You can now login.') 
        return redirect(url_for('auth.login')) 
    return render_template('auth/register.html', form=form)

确认账户

有时需要确认注册时用户提供的信息是否正确，例如能够验证邮箱——需要发送确认邮件，新账户标记为待确认，用户需要点击邮件中包含确认令牌的 URL 链接
确认链接通常为：http://www.example.com/auth/confirm/<确认令牌>

将生成和校验令牌的功能加入 User 模型中

# app/models.py：确认用户账户
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer  # 生成具有过期时间的 JSON Web 签名，类的构造函数参数为一个密钥，可使用SECRET_KEY
from flask import current_app 
from . import db 
 
class User(UserMixin, db.Model): 
    # ... 
    confirmed = db.Column(db.Boolean, default=False) 
 
    def generate_confirmation_token(self, expiration=3600): 
        s = Serializer(current_app.config['SECRET_KEY'], expiration)  # 参数expires_in设置令牌的过期时间（秒）
        return s.dumps({'confirm': self.id})  # dumps()为指定数据生成加密签名并生成序列串
 
    def confirm(self, token): 
        s = Serializer(current_app.config['SECRET_KEY']) 
        try: 
            data = s.loads(token)  # 解码令牌，检验签名和过期时间，如果通过则返回原始数据，或者抛出异常
        except: 
            return False 
        if data.get('confirm') != self.id:  # 检查令牌中的id是否和存储在current_user中的已登录用户匹配
            return False 
        self.confirmed = True 
        db.session.add(self) 
        return True

发送确认邮件

# app/auth/views.py：能发送确认邮件的注册路由
from ..email import send_email 
 
@auth.route('/register', methods = ['GET', 'POST']) 
def register(): 
    form = RegistrationForm() 
    if form.validate_on_submit(): 
        # ... 
        # 重定向之前需要发送确认邮件
        db.session.add(user) 
        db.session.commit()  # 提交数据库之后才能赋予新用户id值
        token = user.generate_confirmation_token()  # 生成令牌
        send_email(user.email, 'Confirm Your Account', 
                   'auth/email/confirm', user=user, token=token)  
        flash('A confirmation email has been sent to you by email.') 
        return redirect(url_for('main.index')) 
    return render_template('auth/register.html', form=form)

邮件需要两个模板：纯文本正文和富文本正文

# app/templates/auth/email/confirm.txt：确认邮件的纯文本正文
Dear {{ user.username }}, 

Welcome to Flasky! 
To confirm your account please click on the following link: 
{{ url_for('auth.confirm', token=token, _external=True) }}  # 一般url_for()生成相对URL，因此需要设置_external=True以生成完整的URL
Sincerely, 
The Flasky Team 
 
Note: replies to this email address are not monitored.

确认账户的视图函数

# 
from flask.ext.login import current_user 
 
@auth.route('/confirm/<token>') 
@login_required 
def confirm(token): 
    if current_user.confirmed:  # 检查已登录的用户是否已经确认过
        return redirect(url_for('main.index')) 
    if current_user.confirm(token):  # 调用user模型的确认功能
        flash('You have confirmed your account. Thanks!') 
    else: 
        flash('The confirmation link is invalid or has expired.') 
    return redirect(url_for('main.index'))

过滤未确认的账户，并显示一个页面，要求用户获取权限前确认账户

# app/auth/views.py：before_app_request中过滤未确认的账户

@auth.before_app_request 
def before_request(): 
    # 同时满足：用户已登录+用户账户没有确认+请求端点不在蓝本中，会拦截请求
    if current_user.is_authenticated() \ 
            and not current_user.confirmed \ 
            and request.endpoint[:5] != 'auth.' \  
            and request.endpoint != 'static':  # endpoint获取请求端点
        return redirect(url_for('auth.unconfirmed'))  # 重定向到路由/auth/unconfirmed
 
@auth.route('/unconfirmed') 
def unconfirmed(): 
    if current_user.is_anonymous() or current_user.confirmed: 
        return redirect(url_for('main.index')) 
    return render_template('auth/unconfirmed.html')

邮件重发

# app/auth/views.py：重新发送账户确认邮件
@auth.route('/confirm') 
@login_required 
def resend_confirmation():  # 重复了一次register()中的操作
    token = current_user.generate_confirmation_token() 
    send_email(current_user.email, 'Confirm Your Account', 
               'auth/email/confirm', user=current_user, token=token) 
    flash('A new confirmation email has been sent to you by email.') 
    return redirect(url_for('main.index'))

管理账户

其他的账户功能：
- 修改密码
- 重设密码（忘记密码）
- 修改电子邮件地址

用户角色

管理员、普通用户、助理等
存在角色的分立，以及权限的组合（这里将用户分为不同角色，但角色用权限定义）

数据库中的表示

Role 模型

# 权限常量
class Permission:  
    FOLLOW = 0x01 
    COMMENT = 0x02 
    WRITE_ARTICLES = 0x04 
    MODERATE_COMMENTS = 0x08 
    ADMINISTER = 0x80

class Role(db.Model): 
    __tablename__ = 'roles' 
    id = db.Column(db.Integer, primary_key=True) 
    name = db.Column(db.String(64), unique=True) 
    default = db.Column(db.Boolean, default=False, index=True) 
    permissions = db.Column(db.Integer)  # 各权限都对应一个位位置，能执行某项操作的角色，其位会被设为1。int一共8位，因此能表示8个权限
    users = db.relationship('User', backref='role', lazy='dynamic')
    
    @staticmethod 
    def insert_roles():  
        # 通过角色名查找现有的角色，然后再进行更新。没有角色名才新建角色对象
        roles = {  # 使用权限组织角色，新增新角色只需使用不同的权限组合即可
            'User': (Permission.FOLLOW | 
                     Permission.COMMENT | 
                     Permission.WRITE_ARTICLES, True), 
            'Moderator': (Permission.FOLLOW | 
                          Permission.COMMENT | 
                          Permission.WRITE_ARTICLES | 
                          Permission.MODERATE_COMMENTS, False), 
            'Administrator': (0xff, False) 
        } 
        for r in roles: 
            role = Role.query.filter_by(name=r).first() 
            if role is None: 
                role = Role(name=r) 
            role.permissions = roles[r][0] 
            role.default = roles[r][1] 
            db.session.add(role) 
        db.session.commit()

角色赋予

用户注册时默认位“一般用户”

管理员注册时，由变量 FLASKY_ADMIN 中的邮箱地址识别，并自动赋予其管理员角色

class User(UserMixin, db.Model): 
    # ... 
    def __init__(self, **kwargs):  
        super(User, self).__init__(**kwargs)  # 构造函数首先调用基类的构造函数，如果没有定义角色，则根据邮箱地址设置角色
        if self.role is None: 
            if self.email == current_app.config['FLASKY_ADMIN']: 
                self.role = Role.query.filter_by(permissions=0xff).first() 
            if self.role is None: 
                self.role = Role.query.filter_by(default=True).first() 
    # ...

角色验证

添加辅助方法，以验证角色是否有指定的权限

# app/models.py
from flask.ext.login import UserMixin, AnonymousUserMixin 
 
class User(UserMixin, db.Model): 
    # ... 
 
    def can(self, permissions):  # 位与操作
        return self.role is not None and \ 
            (self.role.permissions & permissions) == permissions 
 
    def is_administrator(self):  # 单独实现管理员权限检查
        return self.can(Permission.ADMINISTER) 
 
class AnonymousUser(AnonymousUserMixin):  # 匿名角色
    # 设为用户未登录时current_user的值，不用预先检查用户是否登录
    def can(self, permissions): 
        return False 
 
    def is_administrator(self): 
        return False 
 
login_manager.anonymous_user = AnonymousUser

自定义检查用户权限的修饰器

# app/decorators.py
from functools import wraps 
from flask import abort 
from flask.ext.login import current_user 
 
def permission_required(permission):  # 用于常规权限检查
    def decorator(f): 
        @wraps(f) 
        def decorated_function(*args, **kwargs): 
            if not current_user.can(permission): 
                abort(403) 
            return f(*args, **kwargs) 
        return decorated_function 
    return decorator 
 
def admin_required(f):  # 用于管理员权限检查
    return permission_required(Permission.ADMINISTER)(f)

让 Permission 类在所有模板中全局可访问，需要使用上下文处理器，此时 render_template() 可以不用添加 Permission 类的模板参数

# app/main/__init__.py：把 Permission 类加入模板上下文
@main.app_context_processor 
def inject_permissions(): 
    return dict(Permission=Permission)

用户资料

类似个人空间

资料信息

扩充 User 模型，增加用户的其他信息

# app/models.py
class User(UserMixin, db.Model): 
    # ... 
    name = db.Column(db.String(64))  # 用户真实姓名
    location = db.Column(db.String(64)) 
    about_me = db.Column(db.Text())  # db.String需要指定最大长度，db.Text不用
    member_since = db.Column(db.DateTime(), default=datetime.utcnow)  # datetime.utcnow后面没有()， default参数可以接受函数作为默认值
    last_seen = db.Column(db.DateTime(), default=datetime.utcnow)
    
    def ping(self):  # 刷新用户的最后访问时间，此时要在app/auth/views.py的before_request()中增加语句current_user.ping()
        self.last_seen = datetime.utcnow() 
        db.session.add(self)

用户资料页面

为每个用户都创建资料页面

# app/main/views.py：资料页面的路由
@main.route('/user/<username>') 
def user(username): 
    user = User.query.filter_by(username=username).first() 
    if user is None: 
        abort(404) 
    return render_template('user.html', user=user)

设置资料页面的模板

<!--app/templates/user.html：用户资料页面的模板-->
{% block page_content %} 
<div class="page-header"> 
    <h1>{{ user.username }}</h1> 
    {% if user.name or user.location %} 
    <p> 
        <!--name和location字段在同一个<p>元素中，因此至少定义二者的一个时，<p>元素就会创建-->
        {% if user.name %}{{ user.name }}{% endif %} 
        {% if user.location %} 
        <!--location字段被渲染成指向谷歌地图的查询链接-->
            From <a href="http://maps.google.com/?q={{ user.location }}"> 
                {{ user.location }} 
            </a> 
        {% endif %} 
    </p> 
    {% endif %} 
    {% if current_user.is_administrator() %} 
    <!--如果用户是管理员，就显示其邮箱地址-->
    <p><a href="mailto:{{ user.email }}">{{ user.email }}</a></p> 
    {% endif %} 
    {% if user.about_me %}<p>{{ user.about_me }}</p>{% endif %} 
    <p> 
        Member since {{ moment(user.member_since).format('L') }}. 
        Last seen {{ moment(user.last_seen).fromNow() }}. 
    </p> 
</div> 
{% endblock %}

在 base.html 中，在导航条上添加一个链接，指向用户资料页面

资料编辑器

用户要进入一个页面并编辑资料，而管理员也要能够编辑任意用户的资料——包括用户不能直接访问的 User 模型字段

用户级别的资料编辑器

编辑表单

# app/main/forms.py：资料编辑表单
class EditProfileForm(Form):  # 此表单字段都是可选的，因此长度验证函数允许长度为0
    name = StringField('Real name', validators=[Length(0, 64)]) 
    location = StringField('Location', validators=[Length(0, 64)]) 
    about_me = TextAreaField('About me') 
    submit = SubmitField('Submit')

编辑路由

# app/main/views.py：资料编辑路由
@main.route('/edit-profile', methods=['GET', 'POST']) 
@login_required 
def edit_profile(): 
    form = EditProfileForm() 
    if form.validate_on_submit(): 
        current_user.name = form.name.data 
        current_user.location = form.location.data 
        current_user.about_me = form.about_me.data 
        db.session.add(current_user) 
        flash('Your profile has been updated.') 
        return redirect(url_for('.user', username=current_user.username)) 
    form.name.data = current_user.name 
    form.location.data = current_user.location 
    form.about_me.data = current_user.about_me 
    return render_template('edit_profile.html', form=form)

在资料页面添加编辑链接

<!--app/templates/user.html：资料编辑的链接-->
{% if user == current_user %}  <!--只有当用户查看自己的资料页面时才显示-->
<a class="btn btn-default" href="{{ url_for('.edit_profile') }}"> 
    Edit Profile 
</a> 
{% endif %}

管理员级别的资料编辑器