python - 尝试在 flask 上插入时无法解码json对象

Question

我试图在我的flask应用程序上将新用户注册到本地数据库，然后使用psycopg2进行数据库交互。我了解到，使用mogrify可以在db上进行数据插入，但是在服务器端点出现以下错误:

{
  "message": "Failed to decode JSON object: Expecting value: line 1 column 1 (char 0)"
}

我可以从本地数据库检索现有用户，但是无法使用 psycopg2将新用户注册到同一数据库表中。谁能指出我如何进行这项工作？无论如何，我可以将新用户注册/放置到本地数据库表中？有什么想法吗？谢谢

我在 flask 中的尝试:

这是我尝试将新用户注册/放置到本地数据库表的尝试:

from flask_restplus import reqparse, fields, abort
from psycopg2.extensions import AsIs
from flask_restplus import Api, Namespace, Resource, fields, reqparse
from flask import Flask, request, jsonify
import psycopg2, json

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

credential_model = api.model('credential', {
    'username': fields.String(required=True),
    'password': fields.String(required=True)
})

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.expect(credential_parser, validate=True)
    def post(self):
        args = credential_parser.parse_args()

        username = args.get('username')
        password = args.get('password')
        user = {'_id':username,
                'password': password 
        }
        columns = user.keys()
        values = [user[column] for column in columns]
        insert_statement = 'insert into authorized_user_table (%s) values %s'
        try:
            cursor = db.cursor()
            cursor.mogrify(insert_statement, (AsIs(','.join(columns)), tuple(values)))
        except:
            return {"message" : "{} has already been signed".format(username)}, 400
        return { 
            "message" : "{} Register Successfully".format(username), 
            "prediction_id" : username
            }, 200

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

更新:数据库表:

 CREATE TABLE authorized_user_table (
   user_id bigint(20) NOT NULL AUTO_INCREMENT,
   user_name varchar(45) COLLATE utf8_unicode_ci DEFAULT NULL,
   user_email varchar(45) COLLATE utf8_unicode_ci DEFAULT NULL,
   user_password varchar(255) COLLATE utf8_unicode_ci DEFAULT NULL,
   PRIMARY KEY (`user_id`)

insert  into authorized_user_table(user_id,username,email,password) values 
(4,'Jen hardy','jhardy@gmail.com','password4');

为了向数据库表中插入新条目，我使用了 this post和 second post，但是仍然无法将新用户注册到本地数据库表中。

更新:错误仍然存​​在:

我尝试发布答案，但是仍然无法将新用户插入数据库。如何使发布的答案正确？任何想法？

目标:

我只想验证用户是否存在于数据库中(如果不存在)，我想先在本地数据库中注册它们，然后将它们设置为已验证用户可以访问flask应用程序。我该如何进行这项工作？有什么想法吗？谢谢

Answer 1

此代码正常工作。特别是，它使用psycopg2.sql中提供的工具构建INSERT语句，然后执行结果。 cursor.mogrify以字节串形式返回该语句，但不执行该语句。

此代码假定字典保留插入顺序。对于CPython 3.6和Python 3.7的所有实现都是如此。如果使用的是较早版本，则在生成列和值时，必须确保键和值同步。

import sys
import traceback

from flask_restplus import reqparse, fields
from psycopg2 import sql
from flask_restplus import Api, Resource
from flask import Flask
import psycopg2

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

credential_model = api.model(
    "credential",
    {
        "username": fields.String(required=True),
        "password": fields.String(required=True),
    },
)

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.expect(credential_parser, validate=True)
    def post(self):
        args = credential_parser.parse_args()

        username = args.get("username")
        password = args.get("password")
        user = {"user_name": username, "user_password": password}

        insert_statement = "insert into authorized_user_table ({}) values ({})"
        cols = sql.SQL(", ").join([sql.Identifier(x) for x in user.keys()])
        vals = sql.SQL(", ").join([sql.Placeholder() for _ in user.values()])
        insert_statement = sql.SQL(insert_statement).format(cols, vals)

        try:
            cursor = db.cursor()
            cursor.execute(insert_statement, tuple(user.values()))
            # Commit/rollback added to make this work locally 
            # perhaps you are already doing this in your code
            db.commit()
        except: 
            # Print exception traceback if there's an error
            tb = sys.exc_info()
            traceback.print_exception(*tb)
            db.rollback()
            return {"message": "{} has already been signed".format(username)}, 400
        return (
            {
                "message": "{} Register Successfully".format(username),
                "prediction_id": username,
            },
            200,
        )


if __name__ == "__main__":
    db = psycopg2.connect(database="test")
    app.run(debug=True)