Proper way to make a call to an Endpoint from the API using Flask

Question

I'm trying to build a Flask API and I have one endpoint that is supposed to create a user and another one that is suppose to check if a user is present in the database:

@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

The problem is I don't know what is the best way to call my is_present in the new_user endpoint. Should I use requests.get ? Or is there a specific other to do so in Flask?

Answer 1

A common way to handle this is to factor out the view's logic into a separate internal function (i.e. not exposed via the API), and to call the internal function instead. The views will then deal with the request, making a call to the internal function as required. Although it is trivial in this case (it's just a database lookup), this is how it could be done:

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

One obvious advantage of using this method is that the HTTP request is completely avoided, resulting in a more efficient and less error prone (network issues for example) solution.

---

Having said the above as a way of generally dealing with accessing common functionality from multiple views, performing the lookup before creating the new user can not always guarantee that user creation will not fail due to duplicate records. There's a race condition; it is possible for another request to create the user after the first request checks but before it attempts the create operation.

Assuming that your user table has a primary key, e.g. on (first_name, last_name, email), you can simply attempt to create a new user and handle any exception that is raised due to duplicates. Or you might like to look at Session.merge()

Answer 2

I would do something like this for your use-case:

@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