在 for 循环中迭代 fetchall() 字典不起作用

Question

我打算为数据库中的每个用户返回个性化令牌；为此，我尝试在for 循环中迭代用户列表并检查当前用户是否在数据库表中，如果是，GET 方法将返回令牌值。我使用postgresql 和psycopg2，我使用fetchall() 来获取数据库表中的所有实例，但我需要迭代每一行并检查它是否在数据库中。我尝试了 for 循环，但在端点出现以下错误：

curl -X GET --header 'Accept: application/json' 'http://127.0.0.1:5000/token?username=tom%20hardy&password=password3' 

http://127.0.0.1:5000/token?username=tom%20hardy&password=password3

响应正文中的错误消息：

{
  "message": "Username: tom hardy doesn't exist. You have requested this URI [/token] but did you mean /token ?"
}

这不是真的，因为请求正文中的用户实际上是在数据库中。我假设迭代 fetchall() dict 在下面的代码尝试中不起作用。如何使这项工作？

postgresql 数据库：

CREATE TABLE authorized_user_table(
    user_id serial PRIMARY KEY,
    username VARCHAR (50) UNIQUE NOT NULL,
    password VARCHAR (50) NOT NULL
);

insert  into authorized_user_table(user_id,username,password) values 
 (1,'jen hank','password'),
(2,'andy roy','password2'),
(3, 'tom hardy', 'password3'),
(4, 'shah khan', 'password4');

我的后端代码

from flask_restplus import Resource, Api, Namespace
from flask_restplus import abort, fields, inputs, reqparse
from psycopg2 import sql
from flask import Flask, request, jsonify
import psycopg2, json, request

app = Flask(__name__)
api = Api(app) 

credential_parser = reqparse.RequestParser()
credential_parser.add_argument('username', type=str)
credential_parser.add_argument('password', type=str)

@api.route('/token')
class Token(Resource):
    @api.response(200, 'Successful')
    @api.doc(description="Generates a authentication token")
    @api.expect(credential_parser, validate=True)
    def get(self):
        args = credential_parser.parse_args()
        username = args.get('username')
        password = args.get('password')
        cursor = db.cursor()
        cursor.execute('SELECT * FROM public.authorized_user_table')
        users = cursor.fetchall()
        for user in users:
            if username != user[1]:   ## user[1] gives username
                api.abort(404, "Username: {} doesn't exist".format(username))
            if password != user[2]:
                api.abort(401, "Wrong password")
            return {"token": generate_token(username)

def generate_token(self, username):
    info = {
        'username': username,
        'creation_time': time()
    }
    token = self.serializer.dumps(info)
    return token.decode()

if __name__ == '__main__':
    db = psycopg2.connect(database='test_db', user='postgres', password='password', host='localhost', port="5432")
    app.run(debug=True)

在上面的代码中迭代fetchall() dict 仍然不令人满意。我应该如何迭代它们？有什么办法可以做到这一点？谢谢

Answer 1

这个循环中的逻辑不起作用：

    for user in users:
        if username != user[1]:   ## user[1] gives username
            api.abort(404, "Username: {} doesn't exist".format(username))
        if password != user[2]:
            api.abort(401, "Wrong password")

我们正在遍历所有用户，但如果任何用户的名称与请求中的名称不匹配，我们会立即返回 404。同样，如果密码不匹配，我们会立即返回 401。

这样更好：

    ...
    for user in users:
        if username == user[1]:   ## user[1] gives username
            # Assume user names are unique
            # Only check password if the username matches
            if password == user[2]:
                # FIXME: in the question generate_token appears to be 
                # a method of this class, but is outside the class
                return {"token": self.generate_token(username)}
            api.abort(401, "Wrong password")
     api.abort(404, "Username: {} doesn't exist".format(username))

然而，我们可以在单个数据库查询中完成这项工作，方法是询问数据库表中是否存在与请求中的用户名和密码匹配的行。

首先，让我们确保没有两个用户可以拥有相同的用户名，方法是在数据库中唯一：

CREATE UNIQUE INDEX idx_unique_user_name ON authorized_user_table (user_name);

现在在 Python 代码中：

    ...
    # Count the rows that match username AND password
    cursor.execute("""SELECT COUNT(*) FROM public.authorized_user_table """
                   """WHERE user_name = %s AND user_password = %s""",
                   (username, password))
    # COUNT will always return just one row
    user = cursor.fetchone()
    if user[0] == 1:
        return {"token": self.generate_token(username)}
    # Don't let an attacker know what they have got right or wrong.
    api.abort(401, "Invalid user or password")

上述变体在发生错误时返回的信息较少。这对于登录处理程序通常是一个好主意，因为如果攻击者猜测用户名和密码，您不想让他们知道他们是否找到了有效的用户名。

如果您希望响应区分错误的用户名和密码，您可以结合使用这两种方法。

    ...
    cursor.execute("""SELECT user_password FROM public.authorized_user_table """
                   """WHERE user_name = %s""",
                   (username,))
    user = cursor.fetchone()
    if not user:
        api.abort(404, "Username: {} doesn't exist".format(username))
    if user[0] == password:
        return {"token": self.generate_token(username)}
    api.abort(401, "Wrong password")

无论您选择哪种解决方案，要点是在将数据与数据库中的数据进行匹配时，您希望避免从数据库中获取大量行并在应用程序中进行匹配。通过制作合适的查询，让数据库完成工作通常要快得多。