Python Flask MongoDB - Complete CRUD in one video

 

https://www.youtube.com/watch?v=o8jK5enu4L4&ab_channel=CodersPage 

 

Create Python Virtual Environment

python -m venv .

 

Activate Virtual Environment

source /bin/activate

 

Install flask

pip install flask

 

Create server.py in root 

from flask import Flask
app = Flask(__name__)

###################################
@app.route("/users", methods=["POST"])
def create_user():
    return "X"

###################################
if __name__ == '__main__':
    app.run(port=8000, debug=True) #don't use 5000 -> used by control centre

 

Run server

python server.py

 

Setting up postman

1. make a new collection

2. make a new request

 

 

Find Port already in use and kill

lsof -i :5000
kill -9 <PID>

 

Connect to DB

Install pymongo

pip install pymongo

 

 

 

 

Connect DB

locally on localhost:27017

server.py

from flask import Flask, Response 
import pymongo
import json 

app = Flask(__name__)

try:
    mongo = pymongo.MongoClient(
        host="localhost",
        port=27017,
        serverSelectionTimeoutMS = 1000
    )
    db = mongo.company 
    mongo.server_info() # trigger exception if cannot connect to db 
except:
    print("ERROR - Cannot connect to db")

###################################
@app.route("/users", methods=["POST"])
def create_user():
    try:
        user = {"name":"A", "lastName":"AA"}
        dbResponse = db.users.insert_one(user)
        return Response(
            response=json.dumps(
                {"message":"user created",
                 "id": f"{dbResponse.inserted_id}"
                }),
            status=200,
            mimetype="application/json"
        )
    except Exception as e:
        print(e)

###################################
if __name__ == '__main__':
    app.run(port=8000, debug=True)

Test on postman 

 

 

Passing in body

from flask import Flask, Response, request

...

    user = {
        "name":request.form["name"], 
        "lastName":request.form["lastName"]
    }

 

Reading from db 

@app.route("/users", methods=["GET"])
def get_some_users():
    try:
        data = list(db.users.find())
        for user in data:
            user["_id"] = str(user["_id"]) # convert ObjectId to just id
        return Response(
            response=json.dumps(data),
            status=500,
            mimetype="application/json"
        ) 
    except Exception as e:
        print(e)
        return Response(
            response=json.dumps({"message":"cannot read users"}),
            status=500,
            mimetype="application/json"
        )

 

Common error: ObjectId is not serializable 

so here, we need to convert ObjectId to just Id

        for user in data:
            user["_id"] = str(user["_id"]) # convert ObjectId to just id

But later, we will use 

from bson.objectid import ObjectId

 

Update document

@app.route("/users/<id>", methods=["PATCH"]) #can also use PUT
def update_user(id):
    try:
        dbResponse = db.users.update_one(
            {"_id": ObjectId(id)}, # arg1: what we want to update
            {"$set": {"name": request.form["name"]}} # arg2: what we want it to be updated to 
        )
        if dbResponse.modified_count == 1: 
            return Response(
                response=json.dumps({"message":"user updated"}),
                status=200,
                mimetype="application/json"
            )
        else:
            return Response(
                response=json.dumps({"message":"nothing to update"}),
                status=200,
                mimetype="application/json"
            )            
    except Exception as e:
        print(e)
        return Response(
            response=json.dumps({"message":"cannot update"}),
            status=500,
            mimetype="application/json"
        )

 

Check what the response contains 

# inside try 
for attr in dir(dbResponse):
	print(attr)

 

Delete a document

 

@app.route("/users/<id>", methods=["DELETE"])
def delete_user(id):
    try:
        dbResponse = db.users.delete_one({"_id":ObjectId(id)})
        if dbResponse.deleted_count == 1:
            return Response(
                response=json.dumps({"message":"user deleted", "id":f"{id}"}),
                status=200,
                mimetype="application/json"
            )
        else:
            return Response(
                response=json.dumps({"message":"user not found", "id":f"{id}"}),
                status=200,
                mimetype="application/json"
            )
    except Exception as e:
        print(e)
        return Response(
            response=json.dumps({"message":"cannot delete user"}),
            status=500,
            mimetype="application/json"
        )

 

Complete server.py

from flask import Flask, Response, request
import pymongo
import json 
from bson.objectid import ObjectId 

app = Flask(__name__)

# connect to mongo
try:
    mongo = pymongo.MongoClient(
        host="localhost",
        port=27017,
        serverSelectionTimeoutMS = 1000
    )
    db = mongo.company 
    mongo.server_info() # trigger exception if cannot connect to db 
except:
    print("ERROR - Cannot connect to db")


######################################################################
@app.route("/users", methods=["GET"])
def get_some_users():
    try:
        data = list(db.users.find())
        for user in data:
            user["_id"] = str(user["_id"]) # convert ObjectId to just id
        return Response(
            response=json.dumps(data),
            status=500,
            mimetype="application/json"
        ) 
    except Exception as e:
        print(e)
        return Response(
            response=json.dumps({"message":"cannot read users"}),
            status=500,
            mimetype="application/json"
        )

######################################################################
@app.route("/users", methods=["POST"])
def create_user():
    try:
        user = {
            "name":request.form["name"], 
            "lastName":request.form["lastName"]
        }
        dbResponse = db.users.insert_one(user)
        return Response(
            response=json.dumps(
                {"message":"user created",
                 "id": f"{dbResponse.inserted_id}"
                }),
            status=200,
            mimetype="application/json"
        )
    except Exception as e:
        print(e)

######################################################################
@app.route("/users/<id>", methods=["PATCH"]) #can also use PUT
def update_user(id):
    try:
        dbResponse = db.users.update_one(
            {"_id": ObjectId(id)}, # arg1: what we want to update
            {"$set": {"name": request.form["name"]}} # arg2: what we want it to be updated to 
        )
        if dbResponse.modified_count == 1: 
            return Response(
                response=json.dumps({"message":"user updated"}),
                status=200,
                mimetype="application/json"
            )
        else:
            return Response(
                response=json.dumps({"message":"nothing to update"}),
                status=200,
                mimetype="application/json"
            )            
    except Exception as e:
        print(e)
        return Response(
            response=json.dumps({"message":"cannot update"}),
            status=500,
            mimetype="application/json"
        )

######################################################################
@app.route("/users/<id>", methods=["DELETE"])
def delete_user(id):
    try:
        dbResponse = db.users.delete_one({"_id":ObjectId(id)})
        if dbResponse.deleted_count == 1:
            return Response(
                response=json.dumps({"message":"user deleted", "id":f"{id}"}),
                status=200,
                mimetype="application/json"
            )
        else:
            return Response(
                response=json.dumps({"message":"user not found", "id":f"{id}"}),
                status=200,
                mimetype="application/json"
            )
    except Exception as e:
        print(e)
        return Response(
            response=json.dumps({"message":"cannot delete user"}),
            status=500,
            mimetype="application/json"
        )
        
        
        
######################################################################
if __name__ == '__main__':
    app.run(port=8000, debug=True)