1. 程式人生 > >記錄一次完整的flask小型應用開發(2)

記錄一次完整的flask小型應用開發(2)

這一次,我們完成使用者認證的功能:

程式要進行使用者追蹤,程式知道使用者是誰之後,就能針對性的提供體驗。需要使用者提供使用者名稱和密碼。

要是想保證資料庫中存放密碼的安全性,那麼就不存放明文密碼,存放密碼的雜湊值,我們使用Werkzeug來實現密碼雜湊:

所以我們改變models.py中的User模型來支援密碼雜湊

#coding: utf-8
from . import db
from werkzeug.security import generate_password_hash, check_password_hash


class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    users = db.relationship('User', backref='role', lazy='dynamic')

    def __repr__(self):
        return '<Role %r>' % self.name


class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
    password_hash = db.Column(db.String(128))

    @property
    def password(self):
        raise AttributeError('password is not a readable attribute')

    @password.setter
    def password(self, password):
        self.password_hash = generate_password_hash(password)  # 將原始密碼作為輸入,輸出密碼雜湊值


    def verify_password(self, password):
        return check_password_hash(self.password_hash, password)  # 對比資料庫中密碼雜湊值和使用者輸入的密碼,正確返回True

    def __repr__(self):
        return '<User %r>' % self.username

建立認證藍本:

與使用者認證相關的路由可以在auth藍本中定義,對於不同程式功能,我們儘量使用不同的藍本。所以建立app目錄下的auth資料夾,前面建立的main資料夾是主頁基礎功能。
這裡建立藍本的方式與前面差不多:
在auth/init.py中:

from flask import Blueprint

auth = Blueprint('auth', __name__)
from . import views

在auth/views.py中:

from flask import render_template
from . import auth

@auth.route('/login')
def login():
    return render_template('auth/login.html')
# 這裡需要注意了,這個模板檔案需要儲存在auth這個資料夾中
# 但是這個資料夾又需要儲存在app/templates中
# flask認為模板的路徑是相對於程式模板資料夾而言的。

最後需要在create_app()中將藍本註冊到程式上:

from .auth import auth as auth_blueprint
    app.register_blueprint(auth_blueprint, url_prefix='/auth')
    # 這裡加上了prefix,註冊後藍本中定義的所有路由都會加上這個字首
    # 所有views中定義的/login會變成/auth/login

使用Flask-Login來認證使用者:

使用者登入程式後,他們的認證狀態要被記錄下來,這樣瀏覽不同的頁面時才能記住這個狀態。
要想使用Flask-Login,程式User模型必須實現幾個方法,這個拓展提供了一個UserMixin類,包含了方法的預設實現。
修改User模型:

class User(UserMixin, db.Model):  # 下面保持不變,新增一個email欄位

然後我們需要在工廠函式中初始化flask-login:

login_manager = LoginManager()  # 建立一個登入例項
login_manager.session_protection = 'strong'
login_manager.login_view = 'auth.login'  # 設定登入頁面的端點,路由在藍本中定義,所以要加上藍本的名字

最後在create_app()中初始化app:
login_manager.init_app(app)

最後flask-login要求程式使用指定的識別符號載入使用者:這是一個回撥函式

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

然後我們需要建立登入時用到的表單: 因為屬於auth功能,所以寫入auth/forms.py:

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import DataRequired, Length, Email


class LoginForm(FlaskForm):
    email = StringField('Email', validators=[DataRequired(), Length(1, 64),
                                             Email()])
    password = PasswordField('Password', validators=[DataRequired()])
    remember_me = BooleanField('Keep me logged in')
    submit = SubmitField('Log In')

這裡只是表單的功能,我們還需要一個html來展示表單的頁面,我們放在templates/auth/login.html中:

{% extends "/base.html" %}
{% import "bootstrap/wtf.html" as wtf %}

{% block title %}Flasky - Login{% endblock %}

{% block page_content %}
<div class="page-header">
    <h1>Login</h1>
</div>
<div class="col-md-4">
    {{ wtf.quick_form(form) }}
</div>
{% endblock %}

然後我們需要在auth的檢視函式中關聯上表單:

@auth.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()  # 建立一個物件
    # get請求時,檢視函式直接渲染模板顯示錶單
    # POST請求時,拓展的下面這個函式會驗證表單資料
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user is not None and user.verify_password(form.password.data):
            login_user(user, form.remember_me.data)
            next = request.args.get('next')
            if next is None or not next.startswith('/'):
                next = url_for('main.index')
            return redirect(next)
        flash('Invalid username or password.')
    return render_template('auth/login.html', form=form)

然後我們執行,manager.py發現報錯:'A secret key is required to use CSRF.',所以我們需要在config檔案中加上SECRET_KEY = ‘’
結果執行成功!!!!!
然後我們需要一個登出的路由,重定向到首頁:

@auth.route('/logout')
@login_required
def logout():
    logout_user()
    flash('you logged out')
    return redirect(url_for('main.index'))

下面我們實現註冊新使用者的功能:
新增使用者登錄檔單:

class RegistrationForm(FlaskForm):
    email = StringField('Email', validators=[DataRequired(), Length(1, 64),
                                             Email()])
    username = StringField('Username', validators=[
        DataRequired(), Length(1, 64),
        Regexp('^[A-Za-z][A-Za-z0-9_.]*$', 0,
               'Usernames must have only letters, numbers, dots or '
               'underscores')])
    password = PasswordField('Password', validators=[
        DataRequired(), EqualTo('password2', message='Passwords must match.')])
    password2 = PasswordField('Confirm password', validators=[DataRequired()])
    submit = SubmitField('Register')

    def validate_email(self, field):
        if User.query.filter_by(email=field.data).first():
            raise ValidationError('Email already registered.')

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

這還是隻實現了註冊新使用者的功能,我們需要一個註冊新功能的展示頁面:

{% extends "base.html" %}
{% import "bootstrap/wtf.html" as wtf %}

{% block title %}Flasky - Register{% endblock %}

{% block page_content %}
<div class="page-header">
    <h1>Register</h1>
</div>
<div class="col-md-4">
    {{ wtf.quick_form(form) }}
</div>
{% endblock %}

然後我們還需要登入頁面能有一個按鈕能讓使用者跳轉到註冊頁面:

<br>
<p>New user? <a href="{{ url_for('auth.register') }}">Click here to register</a>.</p>

最後別忘了,我們什麼都準備好了,只差完成註冊部分的路由:

@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)
        db.session.commit()
        flash('You can now login.')
        return redirect(url_for('auth.login'))
    return render_template('auth/register.html', form=form)

向用戶郵箱傳送確認註冊郵件:
使用者註冊後,新賬戶首先被標記為待確認狀態,需要使用者按照收到郵件中的說明操作後,才可以證明自己被聯絡上。往往使用者只需要點選一個包含確認令牌的特殊URL:

使用itsdangerous生成確認令牌:
這玩意兒有很多生成令牌的方法,其中TimedJSONWebSignatureSerializer類生成具有過期時間的JSON web簽名

所以我們將生成和檢驗這種令牌的功能新增到User模型:

from itsdangerous import TimedJSONWebSignatureSerializer
from flask import current_app

資料表新增一個欄位:
confirmed = db.Column(db.Boolean, default=False)

# 生成一個令牌,有效期預設一小時
def generate_confirmation_token(self, expiration=3600):
    # 下面生成具有過期時間的JSON web簽名
    s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'], expiration)
    return s.dumps({'confirm': self.id})  # 為指定的資料生成一個加密簽名,然後生成令牌字串

# 檢驗令牌
def confirm(self, token):
    s = TimedJSONWebSignatureSerializer(current_app.config['SECRET_KEY'])
    try:
        data = s.loads(token)  # 這個方法會檢驗簽名和過期時間,如果通過就返回原始資料
    except:
        return False
    if data.get('confirm') != self.id:
        return False
    self.confirmed = True
    db.session.add(self)
    return True

現在令牌已經做出來了,我們需要向用戶傳送確認郵件,之前我們註冊後就直接重定向到index,現在我們需要在這之前,傳送一封確認郵件:

@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)
        db.session.commit()
        # flash('You can now login.')
        # return redirect(url_for('auth.login'))

        # 現在我們要生成令牌然後傳送郵件
        token = user.generate_confirmation()
        send_email(user.email, 'Confirmation of your new account', 'auth/email/confirm', user=user, token=token)
        # 這裡給使用者傳送郵件:收件人,郵件標題,郵件模板,給模板傳的引數
        # 一個電子郵件需要兩個模板,分別用於渲染純文字正文和富文字正文
        flash('we have sent a confirmation email to you, please confirm it!!!')
        return redirect(url_for('main.index'))

    return render_template('auth/register.html', form=form)

郵件模板內容:

<p>Dear {{ user.username }},</p>
<p>Welcome to <b>Flasky</b>!</p>
<p>To confirm your account please <a href="{{ url_for('auth.confirm', token=token, _external=True) }}">click here</a>.</p>
<p>Alternatively, you can paste the following link in your browser's address bar:</p>
<p>{{ url_for('auth.confirm', token=token, _external=True) }}</p>
<p>Sincerely,</p>
<p>The Flasky Team</p>
<p><small>Note: replies to this email address are not monitored.</small></p>

這裡在郵件裡面使用者點選按鈕會開啟網頁,路由為auth.confirm。
所以接下來,我們還需要實現這個路由功能:

# 這個是傳送給使用者的郵件中的路由連結
from flask_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.confirmed(token):
        # 直接呼叫user模型中的驗證令牌方法,直接使用flash來顯示驗證結果。
        flash('you have confirmed your acount!')
    else:
        flash('The confirmation link is invalid or it has expired')
    return redirect(url_for('main.index'))

這裡有一個問題,如果使用者沒有在郵件裡面確認註冊就直接登入,我們可以讓使用者登入到一個頁面,在頁面裡面告訴使用者需要去郵箱裡面確認

這個步驟我們可以使用before_request來完成,但是這個只能應用到屬於藍本的請求上,如果要在藍本中使用針對全域性請求的鉤子,則需要使用before_app_request修飾器:

# 這個部分處理請求前驗證賬號是否被啟用
@auth.before_app_request
def before_request():
    if current_user.is_authenticated \
            and not current_user.confirmed \
            and request.endpoint \
            and request.blueprint != 'auth' \ 
            and request.endpoint != 'static':
        # 需要請求的端點不在認證的藍本當中
        return redirect(url_for('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')

這只是實現了功能,我們還是需要模板:

{% extends "base.html" %}
{% block title %}Flasky - Confirm your account{% endblock %}
{% block page_content %}
<div class="page-header">
    <h1>
        Hello, {{ current_user.username }}!
    </h1>
    <h3>You have not confirmed your account yet.</h3>
    <p>
        Before you can access this site you need to confirm your account.
        Check your inbox, you should have received an email with a confirmation link.
    </p>
    <p>
        Need another confirmation email?
        <a href="{{ url_for('auth.resend_confirmation') }}">Click here</a>
    </p>
</div>
{% endblock %}

可以看到這個模板有一個連結,功能是再次傳送郵件,我們來定義這個路由:

# 這個部分是告訴賬戶未啟用賬戶頁面的連結,用於再次傳送確認郵件
@auth.route('/confirm')
@login_required
def resend_confirmation():
    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'))

這一部分算是徹底完成了,然後我們試著執行一下,發現config中差配置資訊,所以加上:

MAIL_SERVER = os.environ.get('MAIL_SERVER', 'smtp.googlemail.com')
    MAIL_PORT = int(os.environ.get('MAIL_PORT', '587'))
    MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in \
        ['true', 'on', '1']
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
    FLASKY_MAIL_SUBJECT_PREFIX = '[Flasky]'
    FLASKY_MAIL_SENDER = 'Flasky Admin <[email protected]>'
    FLASKY_ADMIN = os.environ.get('FLASKY_ADMIN')
    SQLALCHEMY_TRACK_MODIFICATIONS = False

最後這裡還是有個坑,那就是在設定裡面配置的是gmail,但是國內會出現無網路連線的錯誤,所以可以使用qq郵箱。我們這裡先跳過這個錯誤,大家可在網上自行搜尋如何配置qq郵箱。

使用者角色

因為在web程式中,並不是所有的使用者都有同樣的地位,就比如管理員有著supreme的地位。
首先改進Role模型:

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)

    users = db.relationship('User', backref='role', lazy='dynamic')

    def __repr__(self):
        return '<Role %r>' % self.name

然後列出使用者的許可權,許可權用數字表示

class Permission:
    FOLLOW = 1
    COMMENT = 2
    WRITE = 4
    MODERATE = 8
    ADMIN = 16

現在我們可以建立角色了,但是將角色手動新增到資料庫很麻煩,所以我們在Role類中新增一個方法來完成這個功能。

@staticmethod
    def insert_roles():
        # 這個函式並不是直接建立新的角色,而是通過角色名來查詢現有的角色,再進行更新
        roles = {
            'User': [Permission.FOLLOW, Permission.COMMENT, Permission.WRITE],
            'Moderator': [Permission.FOLLOW, Permission.COMMENT,
                          Permission.WRITE, Permission.MODERATE],
            'Administrator': [Permission.FOLLOW, Permission.COMMENT,
                              Permission.WRITE, Permission.MODERATE,
                              Permission.ADMIN],
        }
        default_role = 'User'
        for r in roles:
            role = Role.query.filter_by(name=r).first()
            if role is None:
                role = Role(name=r)
            role.reset_permissions()
            for perm in roles[r]:
                role.add_permission(perm)
            role.default = (role.name == default_role)
            db.session.add(role)
        db.session.commit()
        
    def add_permission(self, perm):
        if not self.has_permission(perm):
            self.permissions += perm

    def remove_permission(self, perm):
        if self.has_permission(perm):
            self.permissions -= perm

    def reset_permissions(self):
        self.permissions = 0

    def has_permission(self, perm):
        return self.permissions & perm == perm

下面我們給使用者賦予角色,大多數使用者在註冊時候被賦予的角色就是普通使用者,但是當某一個郵箱出現用來註冊的時候,那就直接賦予管理員許可權,所以給User類新增一個初始化方法:

    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(name='Administrator').first()
            if self.role is None:
                self.role = Role.query.filter_by(default=True).first()

下面我們實現角色驗證功能,即特定的路由只能讓擁有特定許可權的使用者才可以使用。
首先我們在User中增加一個方法用來檢查是否有指定的許可權:

    def can(self, perm):
        return self.role is not None and self.role.has_permission(perm)

    def is_administrator(self):
        return self.can(Permission.ADMIN)
class AnonymousUser(AnonymousUserMixin):
    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_login import current_user
from .models import Permission


def permission_required(permission):
    # 用來檢查常規許可權
    def decorator(f):
        @wraps(f)
        # 使用裝飾器時,被裝飾後的函式已經是另外一個函數了,函式名等函式屬性會發生變化
        # 這樣的改變會對測試有所影響,所以這個wraps裝飾器可以消除這樣的副作用。
        def decorated_function(*args, **kwargs):
            if not current_user.can(permission):
                abort(403)  # 使用者沒有許可權就返回403錯誤碼,即HTTP禁止錯誤
            return f(*args, **kwargs)
        return decorated_function
    return decorator


def admin_required(f):
    # 專門用來檢查管理員許可權
    return permission_required(Permission.ADMIN)(f)

那麼如何使用我們剛剛定義的裝飾器呢?

from decorators import admin_requires, permission_required
from .models import Permission

@main.route('/admin')
@login_required
@admin_required
def for_admins_only():
	return 'For administrators!'