The Way to Flask 项目教程：使用 Flask-Login 实现用户注册与登录功能

2025-06-19 06:41:39作者：温玫谨Lighthearted

概述

在现代 Web 应用中，用户认证系统是必不可少的功能模块。本文将详细介绍如何在 Flask 框架中使用 Flask-Login 扩展来实现用户注册、登录和认证功能。这是 The Way to Flask 项目教程系列中的重要章节，将帮助开发者构建安全的用户认证系统。

Flask-Login 简介

Flask-Login 是 Flask 框架的一个扩展，它提供了用户会话管理功能，包括：

用户登录和登出
记住用户会话
保护视图不被未授权用户访问
处理用户会话令牌

环境准备

安装 Flask-Login

首先需要安装 Flask-Login 扩展：

pip install Flask-Login==0.3.2

初始化 Flask-Login

在 Flask 应用中初始化 Flask-Login：

from flask_login import LoginManager

login_manager = LoginManager()
login_manager.init_app(app)

用户模型设计

为了与 Flask-Login 配合使用，我们需要扩展用户模型，实现几个必需的方法：

from flask_mongoengine import MongoEngine

db = MongoEngine()

class User(db.Document):
    name = db.StringField()
    password = db.StringField()
    email = db.StringField()

    def to_json(self):
        return {"name": self.name, "email": self.email}

    # Flask-Login 必需方法
    def is_authenticated(self):
        return True

    def is_active(self):
        return True

    def is_anonymous(self):
        return False

    def get_id(self):
        return str(self.id)

这些方法的作用分别是：

is_authenticated() - 判断用户是否已认证
is_active() - 判断用户是否已激活
is_anonymous() - 判断是否是匿名用户
get_id() - 获取用户唯一标识符

实现用户加载器

Flask-Login 需要通过用户ID加载用户，我们需要实现一个用户加载器：

@login_manager.user_loader
def load_user(user_id):
    return User.objects(id=user_id).first()

登录功能实现

登录视图

from flask_login import login_user

@app.route('/login', methods=['POST'])
def login():
    info = json.loads(request.data)
    username = info.get('username', 'guest')
    password = info.get('password', '')

    user = User.objects(name=username, password=password).first()
    if user:
        login_user(user)  # 关键登录操作
        return jsonify(user.to_json())
    else:
        return jsonify({"status": 401, "reason": "用户名或密码错误"})

设置登录视图

login_manager.login_view = 'login'

登出功能实现

from flask_login import logout_user

@app.route('/logout', methods=['POST'])
def logout():
    logout_user()
    return jsonify(**{'result': 200, 'data': {'message': '登出成功'}})

获取当前用户信息

from flask_login import current_user

@app.route('/user_info', methods=['POST'])
def user_info():
    if current_user.is_authenticated:
        resp = {"result": 200, "data": current_user.to_json()}
    else:
        resp = {"result": 401, "data": {"message": "用户未登录"}}
    return jsonify(**resp)

保护视图

使用 @login_required 装饰器可以保护视图只允许已登录用户访问：

from flask_login import login_required

@app.route('/protected', methods=['GET'])
@login_required
def protected_view():
    return jsonify({"message": "这是受保护的内容"})

完整示例代码

#!/usr/bin/env python
# encoding: utf-8
import json
from flask import Flask, request, jsonify
from flask_login import (current_user, LoginManager,
                         login_user, logout_user,
                         login_required)
from flask_mongoengine import MongoEngine

app = Flask(__name__)
app.config['MONGODB_SETTINGS'] = {
    'db': 'the_way_to_flask',
    'host': 'localhost',
    'port': 27017
}
app.secret_key = 'your-secret-key-here'

db = MongoEngine()
login_manager = LoginManager()
db.init_app(app)
login_manager.init_app(app)
login_manager.login_view = 'login'

# 用户加载器
@login_manager.user_loader
def load_user(user_id):
    return User.objects(id=user_id).first()

# 用户模型
class User(db.Document):
    name = db.StringField()
    password = db.StringField()
    email = db.StringField()

    def to_json(self):
        return {"name": self.name, "email": self.email}

    def is_authenticated(self):
        return True

    def is_active(self):
        return True

    def is_anonymous(self):
        return False

    def get_id(self):
        return str(self.id)

# 登录
@app.route('/login', methods=['POST'])
def login():
    info = json.loads(request.data)
    username = info.get('username', 'guest')
    password = info.get('password', '')

    user = User.objects(name=username, password=password).first()
    if user:
        login_user(user)
        return jsonify(user.to_json())
    else:
        return jsonify({"status": 401, "reason": "用户名或密码错误"})

# 登出
@app.route('/logout', methods=['POST'])
def logout():
    logout_user()
    return jsonify(**{'result': 200, 'data': {'message': '登出成功'}})

# 用户信息
@app.route('/user_info', methods=['POST'])
def user_info():
    if current_user.is_authenticated:
        resp = {"result": 200, "data": current_user.to_json()}
    else:
        resp = {"result": 401, "data": {"message": "用户未登录"}}
    return jsonify(**resp)

# 受保护视图示例
@app.route('/protected', methods=['GET'])
@login_required
def protected_view():
    return jsonify({"message": "这是受保护的内容"})

if __name__ == "__main__":
    app.run(port=8080, debug=True)