6. Flask个人博客实战-用户模块

第一步 先来看下这一节的最终效果 1. clone git项目 并检出到 045df1e7bd0e2cfca5c6342f6d3bec3269df08f8 commit

git clone https://gitee.com/1503319119/flask_blog.git
git reset --hard 5ee990ac9ddd1607337161899f609b2b9120e339
  1. .env 配置mysql数据库

FLASK_ENV=development
DATABASE_URL=mysql+pymysql://数据账号:密码@IP/数据库名称
  1. 安装依赖

pipenv install --dev
  1. 迁移数据库表到mysql

$ pipenv run flask db upgrade
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 2d3d793f6ba4, empty message

5。生成管理员账号

pipenv run flask init admin --username 17305780556 --password 123456
  1. 启动内置服务

pipenv run flask run

7. 访问网站,浏览器打开http://127.0.0.1:5000 输入账号密码就可以登陆了,要是你基础好,直接看代码做对比就可以了。

第二步 实现用户模块 现在回过头来,清空你的测试数据库,删除你的python依赖也就是 .venv 这个目录

git reset --hard 195cd59a967cf6d10ea6f853bd86ba12b6ba0462

在上一个基础上我们来一步一步完成上面的结果: 现在开始对项目用户系统功能分析:

用户登陆
用户退出
用户创建(这里的管理员我们用cli命令行模式创建)
用户列表
用户删除
用户更新

这里肯定需要数据库与一个用户会话管理。这里需要用到的四个扩展:

pipenv install flask-sqlalchemy flask-login pymysql flask-migrate flask-wtf

安装好扩展好,我们在应用中注册扩展

/app/extensions.py
from flask_sqlalchemy import SQLAlchemy as _SQLAlchemy
from flask import current_app, flash
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_wtf import CSRFProtect
from contextlib import contextmanager


class SQLAlchemy(_SQLAlchemy):
   @contextmanager
   def auto_commit(self):
      try:
            yield
            self.session.commit()
      except Exception as e:
            self.session.rollback()
            current_app.logger.error(e)
            flash('未执行成功,请联系管理员', 'danger')
            raise e

db = SQLAlchemy()
migrate = Migrate()
csrf = CSRFProtect()
login_manager = LoginManager()

扩展的配置与引入到时都会在这个文件里引用,这样可以减少循环引入的问题 这里面多了点东西就是注册了一个新的类,继承 SQLAlchemy 增加了一个 auto_commit, 这里主要是用与事务的调用,简化程序。 接着在应用程序APP中注册扩展

app/__init__.py

/app/__init__.py
...
from app.extensions import db, migrate, login_manager, csrf
...


def create_app(config_class=None):
   ...
   register_extensions(app)
   ...


# 注册扩展
def register_extensions(app):
   # 注册jinja的do方法
   app.jinja_env.add_extension('jinja2.ext.do')

   db.init_app(app)
   migrate.init_app(app, db)
   csrf.init_app(app)
   login_manager.init_app(app)

增加一个 register_extensions 方法,并在 create_app 里面调用并把 app 参数注入。 这里就像上一篇的 register_buleprints 一样的逻辑。 接着我们需要几步大家想想,逻辑不分先后, 需要两个蓝图(用户,登陆)这里我分两个,当然用一个也行,但是不方便管理,模型,表单,模板 现在我们先来创建模型。

/app/models/model_mixin.py

/app/models/model_mixin.py
from app.extensions import db


class BaseMixin(object):
   def save(self, form, commit=True, **kwargs):
      """保存数据
      kwargs 里面包含不需要验证直接附值,比如所属于哪个用户
      """
      if form.validate_on_submit() and self._load(form):
            self.__dict__.update(kwargs)
            with db.auto_commit():
               db.session.add(self)
            return self

   def _load(self, form):
      """使用输入数据填充模型
      :params data flaskForm
      """
      for k, v in form.data.items():
            if hasattr(self, k) and k != 'id':
               if v == '':
                  continue

               if form.__dict__[k].type == 'SelectField' and v == 0:
                  setattr(self, k, None)
               else:
                  setattr(self, k, v)
      return True

   def delete(self, physics=False):
      """physics是否是物理删除
      如果是文章模型就设置为回收站
      其它类型删除物理
      """
      with db.auto_commit():
            if not physics:
               self.status = 0
            else:
               db.session.delete(self)
      return True

这里面大概的都说明了,这里看不懂没事,这里主要是对表单的验证与附值做了封装。到时拿着用就行。

/app/models/user.py

/app/models/user.py
from app.extensions import db
from flask_login import UserMixin, AnonymousUserMixin
from datetime import datetime
from app.models.model_mixin import BaseMixin
from app.extensions import login_manager
from werkzeug.security import generate_password_hash, check_password_hash


# User model
class User(db.Model, UserMixin, BaseMixin):
   id = db.Column(db.Integer, primary_key=True)
   username = db.Column(db.String(20))
   nickname = db.Column(db.String(20))
   password_hash = db.Column(db.String(128))
   status = db.Column(db.Integer, default=101)
   last_ip = db.Column(db.String(16))
   last_login_at = db.Column(db.DateTime, default=datetime.now)

   # 这里配置set属性
   @property
   def password(self):
      pass

   # flask_login 他的加载用户所需要的认证函数
   @login_manager.user_loader
   def user_loader(user_id):
      return User.query.get(user_id)

   # 设置密码时,我们用hash存储
   @password.setter
   def password(self, password):
      self.password_hash = generate_password_hash(password)

   # 验证密码
   def validate_password(self, password):
      return check_password_hash(self.password_hash, password)

# 配置flask_login的参数
login_manager.login_view = 'auth.login'
login_manager.login_message = '需要登陆才能访问'
login_manager.login_message_category = 'warning'

创建生成管理员账号命令

/app/cli.py

/app/cli.py
import click
from flask.cli import AppGroup
from app.models.user import User
from app.extensions import db
import random


def register_cli(app):
   init_cli = AppGroup('init', short_help='初始化工作')
   app.cli.add_command(init_cli)

   @init_cli.command(short_help='生成管理员')
   @click.option('--username', default='admin', help='账号')
   @click.option('--password', default='123456', help='密码')
   def admin(username, password):
      if User.query.filter_by(username=username).first():
            raise '账号已经生成,不需要重复生成!'

      with db.auto_commit():
            user = User(username=username, password=password, status=100, nickname='江先生')
            db.session.add(user)
      click.echo('成功生成账号:{}, 密码:{}'.format(username, password))

   @init_cli.command(short_help='重置密码')
   @click.option('--username', help='账号')
   @click.option('--password', help='密码')
   def reset_password(username, password):
      user = User.query.filter(User.username == username).first()
      with db.auto_commit():
            user.password = password

      click.echo('账户密码重置成功')

创建好后,还需要注册到app应用中

/app/__init__.py

/app/__init__.py
from app.cli import register_cli

def create_app(config_class=None):
   ...
   register_cli(app)
   ...

这样,到时就可以在flask 的命令中查看,会多一个init

$ flask
Commands:
db      Perform database migrations.
init    初始化工作
routes  Show the routes for the app.
run     Run a development server.
shell   Runs a shell in the app context.

接着创建表单用来验证,比如用户登陆。创建用户,修改用户。

/app/forms/login.py

/app/forms/login.py
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField
from wtforms.validators import DataRequired, Length


class LoginForm(FlaskForm):
   username = StringField('用户名', validators=[DataRequired(), Length(1, 20)])
   password = PasswordField('密码', validators=[DataRequired(), Length(1, 128)])
   remember_me = BooleanField('记住我')
   submit = SubmitField('登陆')

/app/forms/user.py

/app/forms/user.py
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField, SelectField, ValidationError
from wtforms.validators import DataRequired, Length, Optional, EqualTo
from app.models.user import User
import re


class UserBaseForm(FlaskForm):
   def validate_username(self, field):
      if not re.match(r"^1[35678]\d{9}$", field.data):
            raise ValidationError('必须是手机号码')

      if User.query.filter_by(username=field.data).first():
            raise ValidationError('用户已被占用')

   def load(self, data):
      """填充数据
      """
      for k, v in data.__dict__.items():
            if hasattr(self, k):
               self.__dict__[k].data = '' if v is None else v


# 用户表单
class UserForm(UserBaseForm):
   username = StringField('用户名', validators=[DataRequired('必填'), Length(1, 20)])
   nickname = StringField('昵称', validators=[DataRequired('必填'), Length(1, 20)])
   password = PasswordField('密码', validators=[DataRequired('必填'), Length(6, 128, message='长度6-128个字符'), EqualTo('password2', message="必须与确认密码一致")])
   password2 = PasswordField('确认密码', validators=[DataRequired('必填'), Length(6, 128, message='长度6-128个字符')])
   submit = SubmitField('提交')


# 用户更新
class UserUpdateForm(UserBaseForm):
   nickname = StringField('昵称', validators=[DataRequired('必填'), Length(1, 20)])
   new_password = PasswordField('新密码', validators=[DataRequired(''), Length(6, 128, message='长度6-128个字符'), EqualTo('new_password2', message="必须与确认密码一致")])
   new_password2 = PasswordField('确认新密码', validators=[DataRequired(''), Length(6, 128, message='长度6-128个字符')])
   submit = SubmitField('更新')

就是定义了两个用户的创建与修改表单,这里参数flask_wtf的介绍

接着创建蓝图

/app/admin/auth.py

/app/admin/auth.py
from flask import current_app, Blueprint, render_template, flash, redirect, url_for, request
from app.forms.login import LoginForm
from app.models.user import User
from app.extensions import db
from app.helpers import is_safe_url
from flask_login import login_user, logout_user

bp = Blueprint('auth', __name__)


@bp.route('/login', methods=['GET', 'POST'])
def login():
   form = LoginForm()
   if form.validate_on_submit():
      user = User.query.filter_by(username=form.username.data).first()

      if user is None:
            flash('查不到此用户!', 'warning')
            return redirect(url_for('auth.login'))
      elif not user.validate_password(form.password.data):
            flash('账号或密码错误!', 'warning')
            return redirect(url_for('auth.login'))

      if login_user(user, remember=form.remember_me.data):
            flash('登陆成功!', 'success')
            with db.auto_commit():
               user.last_ip = request.remote_addr
            next_page = request.args.get('next')
            if is_safe_url(next_page):
               return redirect(next_page or url_for('site.index'))
      else:
            flash('账号异常!', 'warning')
            return redirect(url_for('auth.login'))

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


@bp.route('/logout')
def logout():
   logout_user()
   flash('退出成功', 'success')
   return redirect(url_for('auth.login'))

这里用到一个 is_safe_url 来验证跳转的 url

/app/helpers.py

/app/helpers.py
from flask import request
from urllib.parse import urlparse, urljoin

def is_safe_url(target):
   ref_url = urlparse(request.host_url)
   test_url = urlparse(urljoin(request.host_url, target))
   return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc

再创建用户的蓝图

/app/admin/users.py

/app/admin/users.py
from flask import Blueprint, render_template, \
   current_app, request, flash, redirect, url_for
from flask_login import login_required, current_user
from app.models.user import User
from app.forms.user import UserForm, UserUpdateForm
from app.extensions import db

bp = Blueprint('users', __name__)

# 这里不是给这个蓝图的每一次访问的生命周期处验证必须用户登陆才能访问
@bp.before_request
@login_required
def before_request():
   pass

# 用户的列表页
@bp.route('/')
def index():
   page = request.args.get('page', 1, type=int)
   pre_page = current_app.config['POST_PRE_PAGE']
   query = User.query.order_by(User.id.desc())
   pagination = query.paginate(page, pre_page)
   return render_template('admin/users/index.html',
                           model=pagination.items,
                           page=page,
                           pagination=pagination)


# 用户的创建
@bp.route('/create', methods=['GET', 'POST'])
def create():
   form = UserForm()
   model = User()
   if model.save(form):
      flash('添加用户成功', 'success')
      return redirect(url_for('users.index'))
   return render_template('admin/users/create.html', form=form)

# 更新用户
@bp.route('/update/<int:id>', methods=['GET', 'POST'])
def update(id):
   form = UserUpdateForm()
   model = User.query.get_or_404(id)
   if form.validate_on_submit():
      with db.auto_commit():
            if form.nickname.data:
               model.nickname = form.nickname.data
            model.password = form.new_password.data
            flash('更新成功', 'success')
            return redirect(url_for('users.index'))
   form.load(model)
   return render_template('admin/users/update.html', form=form)

# 删除用户
@bp.route('/delete/<int:id>')
def delete(id):
   user = User.query.get_or_404(id)
   if user == current_user:
      flash('不能删除自身!', 'danger')
      return redirect(request.referrer or url_for('user.index'))

   user.delete(True)
   flash('删除成功!', 'success')
   return redirect(url_for('users.index'))

接着创建两个模板(这里就直接跳转到我的码云git上的对应的文件,模板的文章,到时会专门弄一篇文章。

/app/templates/auth
- login.html
/app/templates/users
- index.html
- create.html
- update.html

因为是模板文件有点长,大家直接去 git 上查看 但这里要说一个 flask 分页的模板,这个是通用的

/app/templates/macros/_macros.html

/app/templates/macros/_macros.html
{% macro form_field(field) %}
   <div class="form-group {% if field.errors %}has-error{% endif %}">
      {{ field.label }}
      {{ field(**kwargs) }}
      {% for message in field.errors %}
      <p class="text-danger">{{ message }}</p>
      {% endfor %}
   </div>
{% endmacro %}

{% macro _arg_url_for(endpoint, base) %}
{%- with kargs = base.copy() -%}
{%- do kargs.update(kwargs) -%}
{{url_for(endpoint, **kargs)}}
{%- endwith %}
{%- endmacro %}

{% macro render_pagination(pagination,
                           endpoint=None,
                           prev=('&laquo;')|safe,
                           next=('&raquo;')|safe,
                           size=None,
                           ellipses='…',
                           args={}
                           )
-%}
{% with url_args = {} %}
{%- do url_args.update(request.view_args if not endpoint else {}),
      url_args.update(request.args if not endpoint else {}),
      url_args.update(args) -%}
{% with endpoint = endpoint or request.endpoint %}
<nav>
<ul class="pagination{% if size %} pagination-{{size}}{% endif %} justify-content-end"{{kwargs|xmlattr}}>
{% if prev != None -%}
   <li{% if not pagination.has_prev %} class="page-item disabled"{% endif %}><a class="page-link" href="{{_arg_url_for(endpoint, url_args, page=pagination.prev_num) if pagination.has_prev else '#'}}">{{prev}}</a></li>
{%- endif -%}

{%- for page in pagination.iter_pages() %}
   {% if page %}
      {% if page != pagination.page %}
      <li class="page-item"><a class="page-link" href="{{_arg_url_for(endpoint, url_args, page=page)}}">{{page}}</a></li>
      {% else %}
      <li class="page-item active"><a class="page-link" href="#">{{page}} <span class="sr-only">(current)</span></a></li>
      {% endif %}
   {% elif ellipses != None %}
      <li class="disabled"><a class="page-link" href="#">{{ellipses}}</a></li>
   {% endif %}
{%- endfor %}

{% if next != None -%}
   <li{% if not pagination.has_next %} class="page-item disabled"{% endif %}><a class="page-link" href="{{_arg_url_for(endpoint, url_args, page=pagination.next_num) if pagination.has_next else '#'}}">{{next}}</a></li>
{%- endif -%}
</ul>
</nav>
{% endwith %}
{% endwith %}
{% endmacro %}

这样以后分页,就可以导入,可以参照 /app/templates/admin/user/index.html 里面的使用方法 接着都做好后,我们就要在 app 应用里面把这两个蓝图引入,这样就可以在浏览器里访问了

/app/__init__.py

/app/__init__.py
from app.admin import site, users, auth

# 注册蓝本
def register_buleprints(app):
   # frontend
   app.register_blueprint(site.bp)
   app.register_blueprint(users.bp, url_prefix='/users')
   app.register_blueprint(auth.bp, url_prefix='/auth')

引入模块,并注册蓝本 还需要在配置文件里增加两个配置

/app/config.py

/app/config.py
class Config(object):
   ...
   # flask_sqlalchem 的一个参数
   SQLALCHEMY_TRACK_MODIFICATIONS = False
   # 设置列表页的显示个数,我写4是为了方便测试分页,一般都是选择10,20自由设置
   POST_PRE_PAGE = 4

现在可以看下是否路由注册成功

$ pipenv run flask routes
Loading .env environment variables…
Endpoint      Methods    Rule
------------  ---------  -----------------------
auth.login    GET, POST  /auth/login
auth.logout   GET        /auth/logout
site.index    GET        /
static        GET        /static/<path:filename>
users.create  GET, POST  /users/create
users.delete  GET        /users/delete/<int:id>
users.index   GET        /users/
users.update  GET, POST  /users/update/<int:id>

现在所有的准备都好了,那接下来就是数据库迁移新增加的user表

➜  liubadao_blog git:(master) ✗ pipenv shell
Loading .env environment variables…
...
(liubadao_blog) ➜  liubadao_blog git:(master) ✗ ls
Pipfile      Pipfile.lock app          run.py       wsgi.py
(liubadao_blog) ➜  liubadao_blog git:(master) ✗ flask db init
Creating directory /Users/jydd/workers/test/liubadao_blog/migrations ...  done
...
(liubadao_blog) ➜  liubadao_blog git:(master) ✗ flask db migrate
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'user'
Generating /Users/jydd/workers/test/liubadao_blog/migrations/versions/be23f578f315_.py ...  done
(liubadao_blog) ➜  liubadao_blog git:(master) ✗ flask db upgrade
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> be23f578f315, empty message

数据枯迁移完成后,现在我们用 /app/cli.py 里面的命令来生成一个管理员账号 --password 可自定义密码默认 123456

$ flask init admin --username 17305780556

成功生成账号:17305780552, 密码:123456

总结下流程 | 注册一个cli命令 | 注册两个蓝图,一个登陆,一个用户模块 | 注册一个模型数据user | 创建user的表单,创建与更新创建模板,用到了过滤的东西写到/app/filter.py

接着打开 http://127.0.0.1:5000 就可以查看刚才的功能了。

就可以测试登陆,用户添加,修改的几个功能。