如何解决Flask-Login与MongoDB结合时的用户ID认证问题？

问题描述 #

我在实现登录功能时遇到两个核心问题：

• 无法通过UserMixin的get_id()方法传递正确的user_id
• 不调用User类新实例就无法获取MongoDB生成的默认ID

现有代码 #

models.py #

from flask_login import UserMixin
import bcrypt
from datetime import datetime, UTC
# 假设collection_users是MongoDB的集合对象

class User(UserMixin):
    def __init__(self, email, username, password):
        self.email = email
        self.username = username
        self.password = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
        self.created_at = datetime.now(UTC)

    def to_dict(self):
        return {
            'email': self.email,
            'username': self.username,
            'password': self.password.decode(),
            'created_at': self.created_at
        }

    def save(self):
        user_data = self.to_dict()
        result = collection_users.insert_one(user_data)
        return result.inserted_id

    @classmethod
    def from_dict(cls, data):
        user = cls(data['email'], data['username'], data['password'])
        if 'created_at' in data:
            user.created_at = data['created_at']
        return user

    @classmethod
    def find_by_username(cls, username):
        user_data = collection_users.find_one({'username': username})
        if user_data:
            return user_data
        return None

    @classmethod
    def find_by_email(cls, email):
        email_data = collection_users.find_one({'email': email})
        if email_data:
            return email_data
        return None

    def check_password(self, password):
        try:
            return bcrypt.checkpw(password, self.password)
        except Exception as e:
            print(f"Error checking password: {e}")
            return False

    def is_authenticated(self):
        return self.is_active

    def is_active(self):
        return True

    def is_anonymous(self):
        return False

    # 问题出在这里
    def get_id(self):
        return str(self.to_dict().get('_id'))

routes.py及尝试记录 #

from flask import render_template, redirect, url_for, flash, request
from flask_login import login_user
from .forms import SignInForm
from .models import User
import bcrypt

@auth.route('/login', methods=['GET', 'POST'])
def login():
    form = SignInForm()
    if request.method == 'POST' and form.validate_on_submit():
        username = form.username.data
        password = form.password.data

        found_user = User.find_by_username(username)
        user_id = found_user.get('_id')
        try:
            if found_user and bcrypt.checkpw(password.encode(), found_user['password'].encode()):
                log_user = User(found_user["email"], found_user['username'], found_user["password"])
                # 尝试过的方法及报错：
                # [2] login_user(log_user.get_id(), form.remember_me.data)
                # !> UserMixin.get_id() takes 1 positional arguments but two were given.
                # [3] login_user(log_user, form.remember_me.data)
                # !> No `id` attribute - override `get_id`
                # [4] login_user(user_id, form.remember_me.data)
                # !> 'str' object has no attribute 'is_active'
                # [5] login_user(user_id, form.remember_me.data)
                # !> 'ObjectId' object has no attribute 'is_active'
                # [6] 尝试手动设置属性
                # setattr(log_user, 'is_authenticated', True)
                # !> property 'is_authenticated' of 'User' object has no setter

                login_user(log_user, form.remember_me.data)
                flash(f'Welcome {found_user["username"]}', 'SUCCESS')
                return redirect(url_for('auth.protected'))

        except Exception as e:
            flash(f'Error during login: {str(e)}', 'danger')
            return redirect(url_for('auth.login'))

    return render_template('auth/login.html', title='Sign In', form=form)

问题根源及解决方案 #

核心问题分析 #

1. User实例未存储MongoDB的_id：当前User类的__init__没有接收并保存MongoDB生成的_id，to_dict()也不包含该字段，导致get_id()始终返回None。
2. 查询方法返回字典而非User实例：find_by_username返回的是MongoDB原始字典，不是User对象，后续重新创建的log_user依然没有_id。
3. get_id()实现错误：依赖to_dict()获取_id，但to_dict()本身就没有这个字段。

修复步骤 #

1. 修改User类，添加_id属性并修正相关方法

class User(UserMixin):
    def __init__(self, email, username, password, user_id=None):
        self._id = user_id  # 新增：存储MongoDB的_id
        self.email = email
        self.username = username
        # 区分新用户创建和数据库加载的密码处理
        if isinstance(password, str):
            # 从数据库加载时，密码是哈希字符串，转成bytes
            self.password = password.encode()
        else:
            # 新用户创建时，对明文密码哈希
            self.password = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
        self.created_at = datetime.now(UTC)

    def to_dict(self):
        user_dict = {
            'email': self.email,
            'username': self.username,
            'password': self.password.decode(),
            'created_at': self.created_at
        }
        if self._id is not None:
            user_dict['_id'] = self._id  # 加入_id字段
        return user_dict

    # 直接返回实例的_id字符串，无需依赖to_dict()
    def get_id(self):
        return str(self._id) if self._id else None

2. 修正from_dict方法，正确加载_id

@classmethod
def from_dict(cls, data):
    user_id = data.get('_id')
    # 传入数据库中的哈希密码，避免重复哈希
    user = cls(data['email'], data['username'], data['password'], user_id)
    if 'created_at' in data:
        user.created_at = data['created_at']
    return user

3. 修改查询方法，返回User实例

@classmethod
def find_by_username(cls, username):
    user_data = collection_users.find_one({'username': username})
    if user_data:
        return cls.from_dict(user_data)  # 返回User实例而非字典
    return None

@classmethod
def find_by_email(cls, email):
    email_data = collection_users.find_one({'email': email})
    if email_data:
        return cls.from_dict(email_data)
    return None

4. 简化登录路由代码

@auth.route('/login', methods=['GET', 'POST'])
def login():
    form = SignInForm()
    if request.method == 'POST' and form.validate_on_submit():
        username = form.username.data
        password = form.password.data

        found_user = User.find_by_username(username)
        try:
            # 直接用User实例的check_password方法验证
            if found_user and found_user.check_password(password.encode()):
                login_user(found_user, form.remember_me.data)
                flash(f'Welcome {found_user.username}', 'SUCCESS')
                return redirect(url_for('auth.protected'))
            else:
                flash('用户名或密码错误', 'danger')
        except Exception as e:
            flash(f'登录错误: {str(e)}', 'danger')
            return redirect(url_for('auth.login'))

    return render_template('auth/login.html', title='Sign In', form=form)

额外说明 #

• 无需在登录时重新创建User实例，find_by_username返回的实例已经包含所有数据库字段（包括_id）。
• check_password方法可以直接使用，无需手动调用bcrypt.checkpw。

内容的提问来源于stack exchange，提问作者Feyzullah