Flask Session 登录认证模块

Flask 框架提供了强大的 Session 模块组件，为 Web 应用实现用户注册与登录系统提供了方便的机制。结合 Flask-WTF 表单组件，我们能够轻松地设计出用户友好且具备美观界面的注册和登录页面，使这一功能能够直接应用到我们的项目中。本文将深入探讨如何通过 Flask 和 Flask-WTF 构建一个完整的用户注册与登录系统，以及如何对页面进行优化美化，提高用户体验。通过这一系统，用户能够方便注册账户、安全登录，并且我们能够有效管理用户的会话信息，为 Web 应用的用户管理提供一种高效的解决方案。

什么是Session机制？

Session 是一种在 Web 应用中用于存储用户特定信息的机制。它允许在用户访问网站时存储和检索信息，以便在用户的不同请求之间保持状态。Session 机制在用户登录、购物网站、个性化设置等场景中得到广泛应用，它为用户提供了更加连贯和个性化的体验。在 Flask 中，通过 Flask Session 模块可以方便地使用 Session ，实现用户状态的维护和管理。

在 Web 开发中，HTTP 协议是无状态的，即每个请求都是独立的，服务器不会记住之前的请求信息。为了解决这个问题，引入了 Session 机制。基本思想是在用户访问网站时，服务器生成一个唯一的 Session ID，并将这个 ID 存储在用户的浏览器中（通常通过 Cookie）。同时，服务器端会保存一个映射，将 Session ID 与用户的相关信息关联起来，这样在用户的后续请求中，服务器就能根据 Session ID 找到相应的用户信息，从而实现状态的保持。

Session 的认证流程通常包括以下步骤：

用户登录： 用户通过提供用户名和密码进行登录。在登录验证成功后，服务器为该用户创建一个唯一的 Session ID，并将这个 ID 存储在用户浏览器的 Cookie 中。
Session 存储： 服务器端将用户的相关信息（如用户 ID、权限等）与 Session ID 关联起来，并将这些信息存储在服务器端的 Session 存储中。Session 存储可以是内存、数据库或其他持久化存储方式。
Session ID 传递： 服务器将生成的 Session ID 发送给用户浏览器，通常是通过 Set-Cookie 头部。这个 Cookie 会在用户的每次请求中被包含在 HTTP 头中。
后续请求： 用户在后续的请求中会携带包含 Session ID 的 Cookie。服务器通过解析请求中的 Session ID，从 Session 存储中检索用户的信息，以恢复用户的状态。
认证检查： 服务器在每次请求中检查 Session ID 的有效性，并验证用户的身份。如果 Session ID 无效或过期，用户可能需要重新登录。
用户登出： 当用户主动注销或 Session 过期时，服务器将删除与 Session ID 关联的用户信息，用户需要重新登录。

总体而言，Session 的认证流程通过在客户端和服务器端之间传递唯一的 Session ID，实现了用户状态的持久化和管理。这种机制使得用户可以在多个请求之间保持登录状态，提供了一种有效的用户认证方式。在 Flask 中，开发者可以方便地使用 Flask 提供的 Session 模块来实现这一流程。

Session 认证基础

默认情况下，直接使用Session模块即可实现Session登录会话保持，该方式是将Session存储到内存中，程序重启后即释放，Session的设置一般可以通过使用session["username"]赋值的方式进行，如需验证该Session的可靠性，则只需要调用session.get方法即可一得到特定字段，通过对字段的判断即可实现认证机制。

如下是一个Flask后端代码，运行后通过访问http://127.0.0.1:5000进入到登录这页面。

from flask import Flask,session,render_template,request,Response,redirect,url_for
from functools import wraps
import os

app = Flask(__name__, static_folder="./template",template_folder="./template")
app.config['SECRET_KEY'] = os.urandom(24)

# 登录认证装饰器
def login_required(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        if session.get("username") != None and session.get("is_login") ==True:
            print("登陆过则继续执行原函数")
            return func(*args, **kwargs)
        else:
            print("没有登录则跳转到登录页面")
            resp = Response()
            resp.status_code=200
            resp.data = "<script>window.location.href='/login';</script>"
            return resp
    return wrapper

@app.route("/login",methods=["GET","POST"])
def login():
    if request.method == "GET":
        html = """
                <form action="/login" method="post">
                    <p>账号: <input type="text" name="username"></p>
                    <p>密码: <input type="password" name="password"></p>
                    <input type="submit" value="登录">
                </form>
                """
        return html

    if request.method == "POST":
        get_dict = request.form.to_dict()

        get_username = get_dict['username']
        get_password = get_dict['password']

        if (get_username == "lyshark" or get_username == "admin") and get_password == "123123":
            session["username"] = get_username
            session["is_login"] = True

            print("登录完成直接跳到主页")
            resp = Response()
            resp.status_code=200
            resp.data = "<script>window.location.href='/index';</script>"
            return resp
        else:
            return "登陆失败"
    return "未知错误"

# 主页菜单
@app.route("/index",methods = ["GET","POST"])
@login_required
def index():
    username = session.get("username")
    return "用户 {} 您好,这是主页面".format(username)

# 第二个菜单
@app.route("/get",methods = ["GET","POST"])
@login_required
def get():
    username = session.get("username")
    return "用户 {} 您好,这是子页面".format(username)

@app.route("/logout",methods = ["GET","POST"])
@login_required
def logout():
    username = session.get("username")

    # 登出操作
    session.pop("username")
    session.pop("is_login")
    session.clear()
    return "用户 {} 已注销".format(username)

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

程序运行后，当用户访问http://127.0.0.1:5000地址则会跳转到login登陆页面，此时如果用户第一次访问则会输出如下所示的登陆信息；

通过输入正确的用户名lyshark和密码123123则可以登录成功，此处登录的用户是lyshark如下图。

通过输入不同的用户登录会出现不同的页面提示信息，如下图则是admin的主页信息。

当我们手动输入logout时则此时会退出登录用户，后台也会清除该用户的Session，在开发中可以自动跳转到登出页面；

Session 使用数据库

通过结合 Session 与 SQLite 数据库，我们可以实现一个更完善的用户注册、登录以及密码修改功能。在这个案例中，首先，用户可以通过注册表单输入用户名、密码等信息，这些信息经过验证后将被存储到 SQLite 数据库中。注册成功后，用户可以使用相同的用户名和密码进行登录。登录成功后，我们使用 Flask 的 Session 机制将用户信息保存在服务器端，确保用户在访问其他页面时仍然处于登录状态。

为了增加更多功能，我们还可以实现密码修改的功能。用户在登录状态下，通过密码修改表单输入新的密码，我们将新密码更新到数据库中，确保用户可以安全地更改密码。这个案例综合运用了 Flask、SQLite 和 Session 等功能，为 Web 应用提供了一套完整的用户管理系统。

from flask import Flask,request,render_template,session,Response
import sqlite3,os
from functools import wraps

app = Flask(__name__)

app.config['SECRET_KEY'] = os.urandom(24)

# 创建数据库
def UserDB():
    conn = sqlite3.connect("./database.db")
    cursor = conn.cursor()
    create = "create table UserDB(" \
             "uid INTEGER primary key AUTOINCREMENT not null unique," \
             "username char(64) not null unique," \
             "password char(64) not null," \
             "email char(64) not null" \
             ")"
    cursor.execute(create)
    conn.commit()
    cursor.close()
    conn.close()

# 增删改查简单封装
def RunSqlite(db,table,action,field,value):
    connect = sqlite3.connect(db)
    cursor = connect.cursor()

    # 执行插入动作
    if action == "insert":
        insert = f"insert into {table}({field}) values({value});"
        if insert == None or len(insert) == 0:
            return False
        try:
            cursor.execute(insert)
        except Exception:
            return False

    # 执行更新操作
    elif action == "update":
        update = f"update {table} set {value} where {field};"
        if update == None or len(update) == 0:
            return False
        try:
            cursor.execute(update)
        except Exception:
            return False

    # 执行查询操作
    elif action == "select":

        # 查询条件是否为空
        if value == "none":
            select = f"select {field} from {table};"
        else:
            select = f"select {field} from {table} where {value};"

        try:
            ref = cursor.execute(select)
            ref_data = ref.fetchall()
            connect.commit()
            connect.close()
            return ref_data
        except Exception:
            return False

    # 执行删除操作
    elif action == "delete":
        delete = f"delete from {table} where {field};"
        if delete == None or len(delete) == 0:
            return False
        try:
            cursor.execute(delete)
        except Exception:
            return False
    try:
        connect.commit()
        connect.close()
        return True
    except Exception:
        return False

# 创建数据库
@app.route("/create")
def create():
    UserDB()
    return "create success"

# 登录认证装饰器
def login_required(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        if session.get("username") != None and session.get("is_login") ==True:
            print("登陆过则继续执行原函数")
            return func(*args, **kwargs)
        else:
            print("没有登录则跳转到登录页面")
            resp = Response()
            resp.status_code=200
            resp.data = "<script>window.location.href='/login';</script>"
            return resp
    return wrapper

# 用户注册页面
@app.route("/register",methods=["GET","POST"])
def register():
    if request.method == "GET":
        html = """
                <form action="/register" method="post">
                    <p>账号: <input type="text" name="username"></p>
                    <p>密码: <input type="password" name="password"></p>
                    <p>邮箱: <input type="text", name="email"></p>
                    <input type="submit" value="用户注册">
                </form>
                """
        return html

    if request.method == "POST":
        username = request.form.get("username")
        password = request.form.get("password")
        email = request.form.get("email")

        if RunSqlite("database.db","UserDB","select","username",f"username='{username}'") == []:
            insert = RunSqlite("database.db","UserDB","insert","username,password,email",f"'{username}','{password}','{email}'")
            if insert == True:
                return "创建完成"
            else:
                return "创建失败"
        else:
            return "用户存在"
        return "未知错误"

# 用户登录模块
@app.route("/login",methods=["GET","POST"])
def login():
    if request.method == "GET":
        html = """
                <form action="/login" method="post">
                    <p>账号: <input type="text" name="username"></p>
                    <p>密码: <input type="password" name="password"></p>
                    <input type="submit" value="登录">
                </form>
                """
        return html

    if request.method == "POST":
        username = request.form.get("username")
        password = request.form.get("password")

        select = RunSqlite("database.db","UserDB","select","username,password",f"username='{username}'")
        if select != []:
            # 继续验证密码
            if select[0][1] == password:
                session["username"] = username
                session["is_login"] = True

                print("登录完成直接跳到主页")
                resp = Response()
                resp.status_code = 200
                resp.data = "<script>window.location.href='/index';</script>"
                return resp
            else:
                return "密码不正确"
        else:
            return "用户不存在"
    return "未知错误"

# 修改密码
@app.route("/modify",methods=["GET","POST"])
@login_required
def modify():
    if request.method == "GET":
        html = """
                <form action="/modify" method="post">
                    <p>新密码: <input type="password" name="new_password"></p>
                    <input type="submit" value="修改密码">
                </form>
                """
        return html

    if request.method == "POST":
        username = session.get("username")
        new_password = request.form.get("new_password")
        update = RunSqlite("database.db","UserDB","update",f"username='{username}'",f"password='{new_password}'")
        if update == True:
            # 登出操作
            session.pop("username")
            session.pop("is_login")
            session.clear()

            print("密码已更新,请重新登录")
            resp = Response()
            resp.status_code = 200
            resp.data = "<script>window.location.href='/login';</script>"
            return resp
        else:
            return "密码更新失败"
    return "未知错误"

# 主页菜单
@app.route("/index",methods = ["GET","POST"])
@login_required
def index():
    username = session.get("username")
    return "用户 {} 您好,这是主页面".format(username)

# 第二个菜单
@app.route("/get",methods = ["GET","POST"])
@login_required
def get():
    username = session.get("username")
    return "用户 {} 您好,这是子页面".format(username)

@app.route("/logout",methods = ["GET","POST"])
@login_required
def logout():
    username = session.get("username")

    # 登出操作
    session.pop("username")
    session.pop("is_login")
    session.clear()
    return "用户 {} 已注销".format(username)

if __name__ == '__main__':
    app.run(debug=True)

案例被运行后首先通过调用http://127.0.0.1:5000/create创建database.db数据库，接着我们可以通过访问/register路径实现账号注册功能，如下我们注册lyshark密码是123123，输出效果如下所示；

通过访问/modify可实现对用户密码的修改，但在修改之前需要先通过/login页面登录后进行，否则会默认跳转到用户登录页面中；

使用WTForms登录模板

在如上代码基础上，我们着重增加一个美化登录模板，以提升用户在注册登录流程中的整体体验。通过引入WTF表单组件和Flask-WTF扩展，在前端实现了一个更友好的登录页面。

此登录模板的设计考虑了页面布局、颜色搭配、表单样式等因素，以确保用户在输入用户名和密码时感到轻松自然。同时，我们利用Flask-WTF的验证器功能，对用户输入的数据进行有效性检查，保障了用户信息的安全性。

首先，我们需要在template目录下，创建register.html前端文件，用于用户注册，并写入以下代码。

<html>
<head>
    <link rel="stylesheet" href="https://www.lyshark.com/javascript/bootstrap/3.3.7/css/bootstrap.min.css">
    <link href="https://www.lyshark.com/javascript/font-awesome/4.7.0/css/font-awesome.min.css" rel="stylesheet" type="text/css" />
    <link href="https://www.lyshark.com/javascript/other/my_login.css" rel="stylesheet" type="text/css" />
</head>
<body>
    <div class="container">
        <div class="row">
            <div class="col-md-offset-3 col-md-6">
                <form action="/register" method="post" class="form-horizontal">
                    {{ form.csrf_token }}

                    <span class="heading">用 户 注 册</span>
                    <div class="form-group">
                        {{ form.username }}
                        <i class="fa fa-user"></i>
                        <a href="/login" class="fa fa-question-circle"></a>
                    </div>
                    <div class="form-group">
                        {{ form.email }}
                        <i class="fa fa-envelope"></i>
                    </div>
                    <div class="form-group">
                        {{ form.password }}
                        <i class="fa fa-lock"></i>
                    </div>
                    <div class="form-group">
                        {{ form.RepeatPassword }}
                        <i class="fa fa-unlock-alt"></i>
                    </div>
                    {{ form.submit }}
                </form>
            </div>
        </div>
    </div>
</body>
</html>

接着，继续创建login.html前端文件，用于登录账号时使用，并写入以下代码。

<html>
<head>
    <link rel="stylesheet" href="https://www.lyshark.com/javascript/bootstrap/3.3.7/css/bootstrap.min.css">
    <link href="https://www.lyshark.com/javascript/font-awesome/4.7.0/css/font-awesome.min.css" rel="stylesheet" type="text/css" />
    <link href="https://www.lyshark.com/javascript/other/my_login.css" rel="stylesheet" type="text/css" />
</head>
    <body>
        <div class="container">
            <div class="row">
                <div class="col-md-offset-3 col-md-6">
                    <form action="/login" method="post" class="form-horizontal">
                        {{ form.csrf_token }}

                        <span class="heading">用 户 登 录</span>
                        <div class="form-group">
                            {{ form.username }}
                            <i class="fa fa-user"></i>
                        </div>
                        <div class="form-group help">
                            {{ form.password }}
                            <i class="fa fa-lock"></i>
                            <a href="#" class="fa fa-question-circle"></a>
                        </div>
                        <div class="form-group">
                            <button type="submit" class="btn btn-success">登 录 后 台</button>
                        </div>
                    </form>
                </div>
            </div>
        </div>
    </body>
</html>

后台代码部分，我们需要在原代码的基础之上，增加对前端注册和登录页面的渲染类，此处使用flask_wtf组件实现渲染生成，具体代码如下。

from flask import Flask,request,render_template,session,Response
from functools import wraps
import sqlite3,os

from flask_wtf import FlaskForm
from wtforms import widgets,validators
from wtforms.validators import DataRequired,Regexp,DataRequired, Length, Email, EqualTo, NumberRange
from wtforms.fields import (StringField, PasswordField, DateField, BooleanField,DateTimeField,TimeField,
                            SelectField, SelectMultipleField, TextAreaField,FloatField,HiddenField,
                            RadioField, IntegerField, DecimalField, SubmitField, IntegerRangeField)

# app = Flask(__name__, static_folder="./template",template_folder="./template")
app = Flask(__name__)

app.config["SECRET_KEY"] = "d3d3Lmx5c2hhcmsuY29t"

# -----------------------------------------------------------------------------
# 创建数据库
def UserDB():
    conn = sqlite3.connect("database.db")
    cursor = conn.cursor()
    create = "create table UserDB(" \
             "uid INTEGER primary key AUTOINCREMENT not null unique," \
             "username char(64) not null unique," \
             "password char(64) not null," \
             "email char(64) not null" \
             ")"
    cursor.execute(create)
    conn.commit()
    cursor.close()
    conn.close()

# 增删改查简单封装
def RunSqlite(db,table,action,field,value):
    connect = sqlite3.connect(db)
    cursor = connect.cursor()

    # 执行插入动作
    if action == "insert":
        insert = f"insert into {table}({field}) values({value});"
        if insert == None or len(insert) == 0:
            return False
        try:
            cursor.execute(insert)
        except Exception:
            return False

    # 执行更新操作
    elif action == "update":
        update = f"update {table} set {value} where {field};"
        if update == None or len(update) == 0:
            return False
        try:
            cursor.execute(update)
        except Exception:
            return False

    # 执行查询操作
    elif action == "select":

        # 查询条件是否为空
        if value == "none":
            select = f"select {field} from {table};"
        else:
            select = f"select {field} from {table} where {value};"

        try:
            ref = cursor.execute(select)
            ref_data = ref.fetchall()
            connect.commit()
            connect.close()
            return ref_data
        except Exception:
            return False

    # 执行删除操作
    elif action == "delete":
        delete = f"delete from {table} where {field};"
        if delete == None or len(delete) == 0:
            return False
        try:
            cursor.execute(delete)
        except Exception:
            return False
    try:
        connect.commit()
        connect.close()
        return True
    except Exception:
        return False

# -----------------------------------------------------------------------------
# 生成用户注册表单
class RegisterForm(FlaskForm):
    username = StringField(
        validators=[
            DataRequired(message='用户名不能为空'),
            Length(min=1, max=15, message='用户名长度必须大于%(min)d且小于%(max)d')
        ],
        widget=widgets.TextInput(),
        render_kw={'class': 'form-control', "placeholder":"输入注册用户名"}
    )
    email = StringField(
        validators=[validators.DataRequired(message='邮箱不能为空'),validators.Email(message="邮箱格式输入有误")],
        render_kw={'class':'form-control', "placeholder":"输入Email邮箱"}
    )
    password = PasswordField(
        validators=[
            validators.DataRequired(message='密码不能为空'),
            validators.Length(min=5, message='用户名长度必须大于%(min)d'),
            validators.Regexp(regex="[0-9a-zA-Z]{5,}",message='密码不允许使用特殊字符')
        ],
        widget=widgets.PasswordInput(),
        render_kw={'class': 'form-control', "placeholder":"输入用户密码"}
    )
    RepeatPassword = PasswordField(
        validators=[
            validators.DataRequired(message='密码不能为空'),
            validators.Length(min=5, message='密码长度必须大于%(min)d'),
            validators.Regexp(regex="[0-9a-zA-Z]{5,}",message='密码不允许使用特殊字符'),
            validators.EqualTo("password",message="两次密码输入必须一致")
        ],
        widget=widgets.PasswordInput(),
        render_kw={'class': 'form-control', "placeholder":"再次输入密码"}
    )
    submit = SubmitField(
        label="用 户 注 册", render_kw={ "class":"btn btn-success" }
    )

# 生成用户登录表单
class LoginForm(FlaskForm):
    username = StringField(
        validators=[
            validators.DataRequired(message=''),
            validators.Length(min=4, max=15, message=''),
            validators.Regexp(regex="[0-9a-zA-Z]{4,15}", message='')
        ],
        widget=widgets.TextInput(),
        render_kw={"class":"form-control", "placeholder":"请输入用户名或电子邮件"}
    )
    password = PasswordField(
        validators=[
            validators.DataRequired(message=''),
            validators.Length(min=5, max=15,message=''),
            validators.Regexp(regex="[0-9a-zA-Z]{5,15}",message='')
        ],
        widget=widgets.PasswordInput(),
        render_kw={"class":"form-control", "placeholder":"请输入密码"}
    )

# -----------------------------------------------------------------------------
# 创建数据库
@app.route("/create")
def create():
    UserDB()
    return "create success"

# 登录认证装饰器
def login_required(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        if session.get("username") != None and session.get("is_login") ==True:
            print("登陆过则继续执行原函数")
            return func(*args, **kwargs)
        else:
            print("没有登录则跳转到登录页面")
            resp = Response()
            resp.status_code=200
            resp.data = "<script>window.location.href='/login';</script>"
            return resp
    return wrapper

# 用户注册页面
@app.route("/register",methods=["GET","POST"])
def register():
    form = RegisterForm(csrf_enabled = True)

    if request.method == "POST":
        if form.validate_on_submit():
            username = form.username.data
            password = form.RepeatPassword.data
            email = form.email.data
            print("用户: {} 邮箱: {}".format(username,email))

            if RunSqlite("database.db", "UserDB", "select", "username", f"username='{username}'") == []:
                insert = RunSqlite("database.db", "UserDB", "insert", "username,password,email",
                                   f"'{username}','{password}','{email}'")
                if insert == True:
                    return "创建完成"
                else:
                    return "创建失败"
            else:
                return "用户存在"

    return render_template("register.html", form=form)

# 用户登录页面
@app.route("/login",methods=["GET","POST"])
def login():
    form = LoginForm(csrf_enabled = True)

    if request.method == "POST":
        username = form.username.data
        password = form.password.data

        select = RunSqlite("database.db","UserDB","select","username,password",f"username='{username}'")
        if select != []:
            # 继续验证密码
            if select[0][1] == password:
                session["username"] = username
                session["is_login"] = True

                print("登录完成直接跳到主页")
                resp = Response()
                resp.status_code = 200
                resp.data = "<script>window.location.href='/index';</script>"
                return resp
            else:
                return "密码不正确"
        else:
            return "用户不存在"

    return render_template("login.html", form=form)

# 修改密码
@app.route("/modify",methods=["GET","POST"])
@login_required
def modify():
    if request.method == "GET":
        html = """
                <form action="/modify" method="post">
                    <p>新密码: <input type="password" name="new_password"></p>
                    <input type="submit" value="修改密码">
                </form>
                """
        return html

    if request.method == "POST":
        username = session.get("username")
        new_password = request.form.get("new_password")
        update = RunSqlite("database.db","UserDB","update",f"username='{username}'",f"password='{new_password}'")
        if update == True:
            # 登出操作
            session.pop("username")
            session.pop("is_login")
            session.clear()

            print("密码已更新,请重新登录")
            resp = Response()
            resp.status_code = 200
            resp.data = "<script>window.location.href='/login';</script>"
            return resp
        else:
            return "密码更新失败"
    return "未知错误"

# 主页菜单
@app.route("/index",methods = ["GET","POST"])
@login_required
def index():
    username = session.get("username")
    return "用户 {} 您好,这是主页面".format(username)

# 第二个菜单
@app.route("/get",methods = ["GET","POST"])
@login_required
def get():
    username = session.get("username")
    return "用户 {} 您好,这是子页面".format(username)

@app.route("/logout",methods = ["GET","POST"])
@login_required
def logout():
    username = session.get("username")

    # 登出操作
    session.pop("username")
    session.pop("is_login")
    session.clear()
    return "用户 {} 已注销".format(username)

if __name__ == '__main__':
    app.run(debug=True)