如何在Flask-Login中实现user_loader回调函数

Question

我试图使用Flask和Flask-Login扩展实现Flask应用中的用户认证。目标是从数据库中获取用户账户信息并登录用户，但我遇到了困难； 但是，我已经将其缩小到了Flask-Login行为的特定部分。

根据Flask-Login文档，我需要创建一个user_loader“回调”函数。这个函数的实际目的和实现让我感到困惑了几天：

You will need to provide a user_loader callback. This callback is used to reload the user object from the user ID stored in the session. It should take the Unicode ID of a user, and return the corresponding user object. For example:

@login_manager.user_loader
def load_user(userid):
    return User.get(userid)

现在，假设我想让用户在表单中输入用户名和密码，检查数据库并登录用户。数据库的部分可以正常工作，对我来说没有问题。

这个“回调”函数希望传递一个用户ID＃，并返回用户对象（其内容从数据库中加载）。但我真的不知道它应该检查/执行什么操作，因为用户ID都来自同一个地方。我可以“有点”让回调函数起作用，但它似乎很混乱/不专业，并且它会使浏览器请求的每个资源都访问数据库。我真的不想为了下载favicon.ico而检查我的数据库，但flask-login似乎强制要求这样做。

如果我不再次检查数据库，则无法从此函数返回User对象。 User对象/类在flask登录路由中创建，因此超出了回调的范围。

我无法弄清楚如何将User对象传递到此回调函数中，而不必每次都访问数据库。或者，找出以更有效的方式处理此问题的方法。我一定缺少某些基本知识，但我已经盯着它看了几天，试图使用各种函数和方法，但什么也没用。

这是我测试代码中相关的片段。用户类：

class UserClass(UserMixin):
     def __init__(self, name, id, active=True):
          self.name = name
          self.id = id
          self.active = active

     def is_active(self):
          return self.active

我编写了一个函数，用于返回用户对象到Flask-Login的user_loader回调函数：

def check_db(userid):

     # query database (again), just so we can pass an object to the callback
     db_check = users_collection.find_one({ 'userid' : userid })
     UserObject = UserClass(db_check['username'], userid, active=True)
     if userObject.id == userid:
          return UserObject
     else:
          return None

“回调函数”我并不完全理解（必须返回用户对象，该对象在从数据库中获取后创建）：

@login_manager.user_loader
def load_user(id):
     return check_db(id)

登录路由：

@app.route("/login", methods=["GET", "POST"])
def login():
     if request.method == "POST" and "username" in request.form:
          username = request.form["username"]

          # check MongoDB for the existence of the entered username
          db_result = users_collection.find_one({ 'username' : username })

          result_id = int(db_result['userid'])

          # create User object/instance
          User = UserClass(db_result['username'], result_id, active=True)

          # if username entered matches database, log user in
          if username == db_result['username']:
               # log user in, 
               login_user(User)
               return url_for("index"))
          else:
               flash("Invalid username.")
      else:
           flash(u"Invalid login.")
      return render_template("login.html")

我的代码“有点”能用，我可以登录和退出，但正如我所说，它必须为绝对所有的事情访问数据库，因为我必须在不同的命名空间/范围中向回调函数提供用户对象，而此处发生了其余的登录动作。我非常确定我做错了一切，但是我无法弄清楚怎么做。

Flask-Login提供的示例代码是通过这种方式实现的，但这仅适用于从全局硬编码字典中提取User对象，而不是像在真实世界的场景中那样从数据库中检查并创建用户对象，即在用户输入登录凭据之后。我似乎找不到其他说明使用数据库与Flask-Login的示例代码。

我漏掉了什么？

- Edmond Burnett

---

多奇怪啊，你的login()函数连密码都没检查。 - Houman

@hooman 这只是为了举例而已。 - Edmond Burnett

See - Milovan Tomašević

Answer 1

每次请求都需要从数据库加载用户对象，这一要求的最强理由是Flask-Login将每次检查认证令牌以确保其持续有效性。计算此令牌可能需要在用户对象上存储的参数。

例如，假设一个用户有两个并发会话。在其中一个会话中，用户更改了密码。在随后的请求中，为了使你的应用程序安全，必须注销第二个会话用户并强制重新登录。考虑第二个会话被窃取的情况，因为用户忘记退出计算机-你希望立即通过更改密码来解决问题。你可能还想让管理员有权将用户踢出。

为了实现这种强制注销，存储在cookie中的认证令牌必须1）部分基于密码或其他每次设置新密码时更改的内容；2）在运行任何视图之前针对用户对象的最新已知属性进行检查-这些属性存储在数据库中。

- jd.

---

谢谢你的解释。所以，我的方法/解决方案到目前为止不一定是错误的，而是我在它能够正常工作的原因上有一些错误的假设。 - Edmond Burnett

Answer 2

我非常理解你的担忧，Edmond: 每次需要知道用户角色或名称时都要访问数据库是不可行的。最好的方法是将您的用户对象存储在会话或甚至应用程序范围的缓存中，每隔几分钟从数据库更新一次。我个人使用Redis来实现这一点（这样网站可以在多个线程/进程上运行，同时使用单个缓存入口）。我只需确保Redis配置了密码和非默认端口，并将任何机密数据（例如用户哈希等）以加密形式存储在其中。缓存可以由在指定间隔运行的单独脚本填充，或者在Flask中可以产生单独的线程。注意：Flask-Session也可以配置为使用（相同的）Redis实例存储会话数据，在这种情况下，将需要具有“字节”数据类型的实例，对于常规缓存，您可能经常使用自动将字节转换为字符串的实例类型（decode_responses = True）。

- Anatoly Alekseev

Answer 3

这是我的代码，另一个名为User的数据映射对象提供了query_pwd_md5方法。

用户登录：

@app.route('/users/login', methods=['POST'])
def login():
    # check post.
    uname = request.form.get('user_name')
    request_pwd = request.form.get('password_md5')

    user = User()
    user.id = uname

    try:
        user.check_pwd(request_pwd, BacktestUser.query_pwd_md5(
            uname, DBSessionMaker.get_session()
        ))
        if user.is_authenticated:
            login_user(user)
            LOGGER.info('User login, username: {}'.format(user.id))
            return utils.serialize({'userName': uname}, msg='login success.')
        LOGGER.info('User login failed, username: {}'.format(user.id))
        return abort(401)
    except (MultipleResultsFound, TypeError):
        return abort(401)

用户类：

class User(UserMixin):
"""Flask-login user class.
"""

def __init__(self):
    self.id = None
    self._is_authenticated = False
    self._is_active = True
    self._is_anoymous = False

@property
def is_authenticated(self):
    return self._is_authenticated

@is_authenticated.setter
def is_authenticated(self, val):
    self._is_authenticated = val

@property
def is_active(self):
    return self._is_active

@is_active.setter
def is_active(self, val):
    self._is_active = val

@property
def is_anoymous(self):
    return self._is_anoymous

@is_anoymous.setter
def is_anoymous(self, val):
    self._is_anoymous = val

def check_pwd(self, request_pwd, pwd):
    """Check user request pwd and update authenticate status.

    Args:
        request_pwd: (str)
        pwd: (unicode)
    """
    if request_pwd:
        self.is_authenticated = request_pwd == str(pwd)
    else:
        self.is_authenticated = False

- C .MS