Python SQL 查询类型错误

Question

我想从我的 python 程序中执行 SQL 语句。为此，我使用 MySQLdb 库。这是我的代码：

def execute(sql_statement):
    db = MySQLdb.connect("<DatabaseIP>", "<DatabaseUserName>", "<DatabasePassword>", "<DatabaseName>")
    cursor = db.cursor()
    cursor.execute(sql_statement)
    data = cursor.fetchone()
    db.close()
    return data

def look_up_user_id(username, password):
    print(type(username))
    print(type(password))
    sql_statement = "SELECT ID FROM user WHERE name = '" + username + "' AND password = '" + password + "'"
    print(sql_statement)
    return Database.execute(sql_statement)

这是从请求中提取用户名和密码的方法，也是实际调用数据库class方法的方法。

@app.route('/auth', methods=['GET'])
def log_in():
    return AuthenticationManager.log_in(request.authorization.username, request.authorization.password)

def log_in(username, password):
    return Database.execute(SQLStatementBuilder.look_up_user_id(username, password))[0]

当我执行 print(look_up_user_id("me", "password")) 时，一切正常，我得到了用户的 ID。但是当我在基本身份验证 header 中使用用户名和密码向我的程序发送 HTTP-Request 时，我得到

File "<PythonPath>\Python\Python38\Lib\site-packages\MySQLdb\cursors.py", line 208, in execute
   assert isinstance(query, (bytes, bytearray))
AssertionError

这是引发错误的方法

    def execute(self, query, args=None):
        """Execute a query.

        query -- string, query to execute on server
        args -- optional sequence or mapping, parameters to use with query.

        Note: If args is a sequence, then %s must be used as the
        parameter placeholder in the query. If a mapping is used,
        %(key)s must be used as the placeholder.

        Returns integer represents rows affected, if any
        """
        while self.nextset():
            pass
        db = self._get_db()

        if isinstance(query, unicode):
            query = query.encode(db.encoding)

        if args is not None:
            if isinstance(args, dict):
                nargs = {}
                for key, item in args.items():
                    if isinstance(key, unicode):
                        key = key.encode(db.encoding)
                    nargs[key] = db.literal(item)
                args = nargs
            else:
                args = tuple(map(db.literal, args))
            try:
                query = query % args
            except TypeError as m:
                raise ProgrammingError(str(m))
        assert isinstance(query, (bytes, bytearray))
        res = self._query(query)
        return res

方法 look_up_user_id 中的两次打印调用只是 return <class 'str'> 在两个测试用例中以及密码和名称。

感谢您的帮助！

编辑：添加了引发错误的方法

Answer 1

问题是 lookup_user_id 正在 returning Database.execute(sql_statement) 的结果并且结果再次传递给 Database.execute 在 AuthorizationManager 的 return 语句中。要解决此问题，请使用 SQLStatementBuilder return sql_statement。另请注意，log_in 视图应 return 字符串或类似字符串。

另外，考虑像这样执行 SQL 语句：

res = cursor.execute("""SELECT id FROM user WHERE name = %s and password = %s""",
                     (username, password))

防止SQL注入攻击。

Python SQL 查询类型错误

Python SQL query has the wrong type

python

database

types

database-connection

assertion