如何使用Flask中的寄存器？

Question

我有以下app.py：

from flask import Flask
from waitress import serve
from bprint import api_blueprint
from errors import invalid_id, not_found, invalid_input, internal_server_error, unauthorized_access

app = Flask(__name__)

app.register_blueprint(api_blueprint)

app.register_error_handler(400, invalid_id)
app.register_error_handler(401, unauthorized_access)
app.register_error_handler(404, not_found)
app.register_error_handler(405, invalid_input)
app.register_error_handler(500, internal_server_error)

if __name__ == "__main__":
    serve(app, host='localhost')

以及bprint.py中的以下代码：

from flask import Blueprint, jsonify, request
import dbu
from models import Session, user_table, car_table, order_table
from schema import UserDetails, UserQuery, OrderDetails, OrderQuery, CarDetails, CarQuery, LoginData, \
    ListUsersReq, Response
from contextlib import contextmanager
from flask_jwt_extended import jwt_required, create_access_token, get_jwt_identity
import datetime

api_blueprint = Blueprint('api', __name__)


@contextmanager
def session_scope():
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    else:
        try:
            session.commit()
        except:
            session.rollback()
            raise


@api_blueprint.route("/login", methods=["POST"])
def login():
    from app import bcrypt
    data = LoginData().load(request.json)
    if data:
        user = dbu.get_entry_by_username(user_table, username=data["username"])
        hpw = bcrypt.generate_password_hash(data["password"])
        if not user:
            return jsonify({"message": "Couldn't find user!"})
        if bcrypt.check_password_hash(hpw, data["password"]):
            access_token = create_access_token(identity=data["username"], expires_delta=datetime.timedelta(days=365))
            return jsonify(access_token=access_token, id=user.id), 200


@api_blueprint.route("/user", methods=["GET"])
def list_users():
    with session_scope():
        current_user = get_jwt_identity()
        user = dbu.get_entry_by_username(user_table, current_user)
        if user.admin:
            args = ListUsersReq().load(request.args)
            userlist = dbu.list_users(args.get("email"), args.get("username"))
            return jsonify(UserDetails(many=True).dump(userlist))
        else:
            return jsonify(code=401, type='UNAUTHORIZED_ACCESS'), 401


@api_blueprint.route("/user", methods=["POST"])
def create_user():
    with session_scope():
        from app import bcrypt
        user_details = UserQuery().load(request.get_json(force=True))
        user_details["password"] = bcrypt.generate_password_hash(user_details["password"]).decode('UTF-8')
        user = dbu.create_entry(user_table, **user_details)
        access_token = create_access_token(identity=user.username, expires_delta=datetime.timedelta(days=365))
        return jsonify(access_token=access_token, id=UserDetails().dump(user)["id"]), 200


@api_blueprint.route("/user/<int:id>", methods=["GET"])
def user_by_id(id):
    with session_scope():
        current_user = get_jwt_identity()
        user = dbu.get_entry_by_username(user_table, current_user)
        if user.admin:
            user = dbu.get_entry_by_id(user_table, id)
            return jsonify(UserDetails().dump(user))
        else:
            return jsonify(code=401, type="UNAUTHORIZED_ACCESS"), 401


@api_blueprint.route("/user/<int:id>", methods=["PUT"])
def update_user(id):
    with session_scope():
        current_user = get_jwt_identity()
        user = dbu.get_entry_by_username(user_table, current_user)
        if user.admin or user.id == id:
            user_details = UserQuery().load(request.json)
            user = dbu.get_entry_by_id(user_table, id)
            dbu.update_entry(user, **user_details)
            return jsonify(Response().dump({"code": "200"}))
        else:
            return jsonify(code=401, type="UNAUTHORIZED_ACCESS"), 401


@api_blueprint.route("/user/<int:id>", methods=["DELETE"])
def delete_user(id):
    with session_scope():
        current_user = get_jwt_identity()
        user = dbu.get_entry_by_username(user_table, current_user)
        if user.admin or user.id == id:
            dbu.delete_entry(user_table, id)
            return jsonify(Response().dump({"code": "200"}))
        else:
            return jsonify(code=401, type="UNAUTHORIZED_ACCESS"), 401


@api_blueprint.route("/cars", methods=["GET"])
def get_inventory():
    with session_scope():
        cars = dbu.list_cars()
        return jsonify(CarDetails(many=True).dump(cars))


@api_blueprint.route("/cars/car/<int:carId>", methods=["GET"])
def get_car_by_id(carId):
    with session_scope():
        car = dbu.get_car_by_id(car_table, carId)
        return jsonify(CarDetails().dump(car))


@api_blueprint.route("/cars/car", methods=["POST"])
def create_car():
    with session_scope():
        current_user = get_jwt_identity()
        user = dbu.get_entry_by_username(user_table, current_user)
        if user.admin:
            car_details = CarQuery().load(request.json)
            car = dbu.create_entry(car_table, **car_details)
            return jsonify({"carId": CarDetails().dump(car)["carId"]})
        else:
            return jsonify(code=401, type="UNAUTHORIZED_ACCESS"), 401


@api_blueprint.route("/cars/car/<int:carId>", methods=["PUT"])
def update_car(carId):
    with session_scope():
        current_user = get_jwt_identity()
        user = dbu.get_entry_by_username(user_table, current_user)
        if user.admin:
            car_details = CarQuery().load(request.json)
            car = dbu.get_car_by_id(car_table, carId)
            dbu.update_entry(car, **car_details)
            return jsonify(Response().dump({"code": "200"}))
        else:
            return jsonify(code=401, type="UNAUTHORIZED_ACCESS"), 401


@api_blueprint.route("/cars/car/<int:carId>", methods=["DELETE"])
def delete_car(carId):
    with session_scope():
        current_user = get_jwt_identity()
        user = dbu.get_entry_by_username(user_table, current_user)
        if user.admin:
            dbu.delete_car(car_table, carId)
            return jsonify(Response().dump({"code": "200"}))
        else:
            return jsonify(code=401, type="UNAUTHORIZED_ACCESS"), 401


@api_blueprint.route("/cars/car/<int:carId>/order", methods=["POST"])
def place_order(carId):
    with session_scope():
        current_user = get_jwt_identity()
        user = dbu.get_entry_by_username(user_table, current_user)
        if user:
            order_data = OrderQuery().load(request.json)
            order = dbu.create_entry(order_table,
                                     userId=user.id,
                                     carId=carId,
                                     shipDate=order_data["shipDate"],
                                     returnDate=order_data["returnDate"],
                                     status="placed",
                                     complete=False)
            return jsonify({"id": OrderDetails().dump(order)["id"]})
        else:
            return jsonify(code=401, type="UNAUTHORIZED_ACCESS"), 401


@api_blueprint.route("/orders", methods=["GET"])
def get_orders():
    with session_scope():
        current_user = get_jwt_identity()
        user = dbu.get_entry_by_username(user_table, current_user)
        if user.admin:
            orders = dbu.list_orders()
            return jsonify(OrderDetails(many=True).dump(orders))
        else:
            return jsonify(code=401, type="UNAUTHORIZED_ACCESS"), 401


@api_blueprint.route("/cars/car/<int:carId>/order/<int:orderId>", methods=["GET"])
def get_order_by_id(carId, orderId):
    with session_scope():
        current_user = get_jwt_identity()
        user = dbu.get_entry_by_username(user_table, current_user)
        if user:
            order = dbu.get_entry_by_id(order_table, id=orderId)
            return jsonify(OrderDetails().dump(order))
        else:
            return jsonify(code=401, type="UNAUTHORIZED_ACCESS"), 401


@api_blueprint.route("/cars/car/<int:carId>/order/<int:orderId>", methods=["DELETE"])
def delete_order(carId, orderId):
    with session_scope():
        dbu.delete_entry(order_table, id=orderId)
        return jsonify(Response().dump({"code": "200"}))

当我尝试运行waitress-serve --port=5000 app:app时，我得到以下错误：

    app.register_blueprint(api_blueprint)
    raise AssertionError(
AssertionError: View function mapping is overwriting an existing endpoint function: api.wrapper

有什么问题吗？ 我几乎可以肯定它在12月份起作用了，现在在重新安装Windows之后就不起作用了 现在它还说我的问题中有太多的代码，我不知道如何用更多的词来解释我的问题，所以我不得不添加一些无用的行，对不起

Answer 1

检查视图上是否有一些自定义装饰器。因为flask从@route参数或函数名中获取端点名称。在您的例子中，任何函数中都没有endpoint参数。 错误显示为api.wrapper，这意味着您有两个或更多名为wrapper的函数。通常我们在装饰师里面看到这样的名字。所以你可能有一个看起来像

def decorator(f):
    def wrapper(*args, **kwargs):
        return f(*args, **kwargs)
    return wrapper

有些景观是用这种装饰物装饰的。并将“wrapper”作为视图名称

Answer 2

试着注释几行代码——这可能会帮助您解决问题

from flask import Flask
from waitress import serve
from bprint import api_blueprint
# from errors import invalid_id, not_found, invalid_input, internal_server_error, unauthorized_access

app = Flask(__name__)

app.register_blueprint(api_blueprint)

# app.register_error_handler(400, invalid_id)
# app.register_error_handler(401, unauthorized_access)
# app.register_error_handler(404, not_found)
# app.register_error_handler(405, invalid_input)
# app.register_error_handler(500, internal_server_error)

if __name__ == "__main__":
    serve(app, host='localhost')