使用 Flask 从 API 调用端点的正确方法

Question

我正在尝试构建一个 Flask API，我有一个端点应该创建一个用户，另一个端点应该检查用户是否存在于数据库中：

@API.route('/users/', methods=['POST'])
def new_user():
    user_json = json.loads(request.get_json())
    first_name = user_json.get('first_name')
    last_name = user_json.get('last_name')
    email = user_json.get('email')
    password = user_json.get('password')
    # Call the other endpoint here
    if response == 400:
        try:
           user = User(first_name=first_name, last_name=last_name, email=email, password=password)
           db.session.add(user)
           db.session.commit()
           return jsonify(user=user.to_json()), 200
       except:
           return jsonify(error=500), 500
    else:
       return jsonify(user=user.to_json()), 409



@API.route('/users/<string:email>', methods=['GET'])
def is_present(email):
    user = User.query.filter_by(email=email).first()
    if user:
        print(user)
        return jsonify(user=user.to_json()), 200
    else:
        return jsonify(error=404), 404

问题是我不知道在new_user 端点中调用我的is_present 的最佳方式是什么。我应该使用requests.get 吗？或者在 Flask 中是否有特定的其他人可以这样做？

Answer 1

处理此问题的常用方法是将视图的逻辑分解为单独的内部函数（即不通过 API 公开），并改为调用内部函数。然后视图将处理请求，根据需要调用内部函数。尽管在这种情况下它是微不足道的（它只是一个数据库查找），但可以这样做：

def get_user(email):
    return User.query.filter_by(email=email).first()

@API.route('/users/', methods=['POST'])
def new_user():
    user_json = json.loads(request.get_json())
    first_name = user_json.get('first_name')
    last_name = user_json.get('last_name')
    email = user_json.get('email')
    password = user_json.get('password')
    user = get_user(email)
    if not user:
        try:
           user = User(first_name=first_name, last_name=last_name, email=email, password=password)
           db.session.add(user)
           db.session.commit()
           return jsonify(user=user.to_json()), 200
       except:
           return jsonify(error=500), 500
    else:
       return jsonify(user=user.to_json()), 409

@API.route('/users/<string:email>', methods=['GET'])
def is_present(email):
    user = get_user(email)
    if user:
        print(user)
        return jsonify(user=user.to_json()), 200
    else:
        return jsonify(error=404), 404

使用此方法的一个明显优势是完全避免了 HTTP 请求，从而获得了更高效且不易出错（例如网络问题）的解决方案。

---

以上所说的通常是处理从多个视图访问公共功能的一种方式，但在创建新用户之前执行查找并不能始终保证用户创建不会由于重复记录而失败。有一个竞争条件；在第一次请求检查之后但在尝试创建操作之前，另一个请求可能会创建用户。

假设您的用户表有一个主键，例如在(first_name, last_name, email) 上，您可以简单地尝试创建一个新用户并处理由于重复而引发的任何异常。或者你可能想看看Session.merge()

Answer 2

我会为您的用例做这样的事情：

@API.route('/users', defaults={'email': None} ,methods=['GET', 'POST'])
@API.route('/users/<string:email>', methods=['GET', 'POST'])
def new_user(email):
    if(email):
        user = User.query.filter_by(email=email).first()
        if user:
            return jsonify(user=user.to_json()), 200
        else:
            return jsonify(error=404), 404
    else:
        user_json = json.loads(request.get_json())
        first_name = user_json.get('first_name')
        last_name = user_json.get('last_name')
        email = user_json.get('email')
        password = user_json.get('password')
        user = User(first_name=first_name, last_name=last_name, email=email, password=password)
        db.session.add(user)
        db.session.commit()
        return jsonify(user=user.to_json()), 200