Python Rest API – Database

In this article we will see how to integrate database specially mongodb with our api.

Till now all our database has been stored in lists, but instead we need store data in database now.

I use mongodb since its very fast to write code on it, no need to setup db structure etc, just start to write code.

We will use https://flask-pymongo.readthedocs.io/en/latest/ , so install it and do the basic setup as per the documentation.

$ pip install Flask-PyMongo 

 

from flask_pymongo import PyMongo

app = Flask(__name__)
app.config["MONGO_URI"] = "mongodb://localhost:27017/todo"
mongo = PyMongo(app) 

In this post we will just do simple CRUD operation of our todo app in mongo

P.S: I am making login optional in CRUD routes just for the sake of testing code faster.

TODO CReate

 
@app.route('/todo', methods=["POST"])
@jwt_optional
def add_todo():
    if not request.json:
        abort(500)

    title = request.json.get("title", None)
    desc = request.json.get("description", "")

    due = request.json.get("due", None)

    if due is not None:
        due = datetime.datetime.strptime(due, "%d-%m-%Y")
    else:
        due = datetime.datetime.now()

    current_user = get_current_user()

    user_id = ""
    if current_user is not None and "id" in current_user:
        user_id = current_user["id"]

    # use insert_one to do insert operations. id is auto generated    
    ret = mongo.db.tasks.insert_one({
        "title": title,
        "description": desc,
        "done": False,
        "due": due,
        "user":  user_id
    }).inserted_id

    # fetch the inserted id and convert it to string before sending it in response
    return jsonify(str(ret)) 

See inline code comments for explanation

Todo Update

Let’s see update

 
@app.route("/todo/<string:id>", methods=['PUT'])
@jwt_optional
def update_todo(id):

    if not request.json:
        abort(500)

    title = request.json.get("title", None)
    desc = request.json.get("description", "")

    if title is None:
        return jsonify(message="Invalid Request"), 500

    update_json = {}
    if title is not None:
        update_json["title"] = title

    if desc is not None:
        update_json["description"] = desc


    # match with Object ID
    ret = mongo.db.tasks.update({
        "_id": ObjectId(id)
    }, {
        "$set": update_json
    }, upsert=False)

    return jsonify(ret) 

Todo Read

 
@app.route('/todo', methods=["GET"])
@app.route('/todo/<string:direction>', methods=["GET"])
@jwt_optional
def todo(direction=None):
    # direction is optional
    current_user = get_current_user()
    if direction == "ASC":
        direction = 1
    else:
        direction = -1

    if direction is not None:
        if current_user is not None and "id" in current_user:
            if current_user["role"] == "normal":
                ret = mongo.db.tasks.find(
                    {"user": current_user["id"]}).sort("due", direction)
        else:
            ret = mongo.db.tasks.find({"user": ""}).sort(
                "due", direction).limit(5)
    else:
        if current_user is not None and "id" in current_user:
            ret = mongo.db.tasks.find(
                {"user": current_user["id"]})
        else:
            ret = mongo.db.tasks.find({"user": ""}).limit(5)

    tasks = []
    for doc in ret:
        doc["_id"] = str(doc["_id"])
        tasks.append(doc)
    return jsonify(tasks) 

Todo Delete

Finally, delete

 
@app.route("/todo/<string:id>", methods=["DELETE"])
def delete_todo(id):

    ret = mongo.db.tasks.remove({
        "_id" : ObjectId(id)
    })

    return jsonify(ret) 

So we have complete full CRUD operations using mongodb and show how easy is to implement mongo with python!

Next, you could implement the full todo app with authorization on mongo to improve your skills.