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

Posted 2023-02-23

【中文标题】使用 Flask 从 API 调用端点的正确方法

【英文标题】：Proper way to make a call to an Endpoint from the API using Flask

【发布时间】：2017-09-04 11:07:44

【问题描述】：

我正在尝试构建一个 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 中是否有特定的其他人可以这样做？

【参考方案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()

【讨论】：

好的，谢谢你的提示！我想正确编码，我应该把像'get_user'这样的函数放在我的存储库中的哪里？如果我还有一个端点需要访问另一个端点，我应该怎么做？

如果您坚持发出 HTTP 请求，requests 是一个不错的选择。 Flask 确实提供了test_client()，所以如果你能克服它被命名为“测试”客户端而不是通用 HTTP 客户端，你可以使用它。

在哪里放置get_user()？如果你有一个模型文件，它可以去那里，否则 User 类被声明的地方似乎是最好的地方。

【参考方案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