Upload server.py

server.py

+from flask import Flask, request
+import random
+import json
+import database_helper as database_helper
+import re
+
+app = Flask(__name__)
+
+
+# function to signin the user with post method
+# We will expect 2 string values email address password.
+# if the authentication is successful we will return access token
+@app.route("/sign_in", methods=["POST"])
+def sign_in():
+    email_address = request.form.get("email", "")
+    password = request.form.get("password", "")
+
+    if len(email_address) == 0 or len(password) == 0:
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        )
+
+    result = database_helper.find_user_by_email_password(email_address, password)
+
+    if result["status"] == True:
+        if len(result["data"]) == 0:
+            return json.dumps(
+                {"success": False, "message": "Wrong username or password."}
+            )
+        else:
+            token = random.getrandbits(30)
+            database_helper.add_user_email_token(email_address, token)
+            return json.dumps(
+                {
+                    "success": True,
+                    "message": "Successfully sign in.",
+                    "data": str(token),
+                }
+            )
+
+    return json.dumps({"success": False, "message": "error in sign in"})
+
+
+# function to signup the user with post method
+# We will expect 7 string values following: email, password, firstname, familyname, gender, city and country.
+# if the sign_up is successful we will return access token
+@app.route("/sign_up", methods=["POST"])
+def sign_up():
+    regex = r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b"
+    email_address = request.form.get("email", "")
+    password = request.form.get("password", "")
+    firstname = request.form.get("firstname", "")
+    familyname = request.form.get("familyname", "")
+    gender = request.form.get("gender", "")
+    city = request.form.get("city", "")
+    country = request.form.get("country", "")
+
+    if (
+        len(email_address) == 0
+        or len(password) == 0
+        or len(firstname) == 0
+        or len(familyname) == 0
+        or len(gender) == 0
+        or len(city) == 0
+        or len(country) == 0
+    ):
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        )
+
+    if len(password) < 8:
+        return json.dumps(
+            {
+                "success": False,
+                "message": "Password should be at least 8 characters long.",
+            }
+        )
+
+    if re.fullmatch(regex, email_address) is None:
+        return json.dumps(
+            {
+                "success": False,
+                "message": "Invalid email, please provide valid email address",
+            }
+        )
+
+    if database_helper.user_exists_email(email_address):
+        return json.dumps({"success": False, "message": "User already exists."})
+
+    result = database_helper.create_user(
+        email_address, password, firstname, familyname, gender, city, country
+    )
+
+    if result["status"] == True:
+        return json.dumps(
+            {"success": True, "message": "Successfully created a new user."}
+        )
+
+    return json.dumps({"success": False, "message": "error in sign up"})
+
+
+# function to signout the user with post method
+# We will expect 1 string values token.
+# if the token is valid we will signout the user
+@app.route("/sign_out", methods=["DELETE"])
+def sign_out():
+    token = request.headers.get("token", "")
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."})
+
+    result = database_helper.find_token(token)
+
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not signed in."})
+
+    result = database_helper.remove_token(token)
+
+    if result["status"] == True:
+        return json.dumps({"success": True, "message": "Successfully signed out."})
+    elif result["status"] == False:
+        return json.dumps({"success": False, "message": result["data"]})
+
+    return json.dumps({"success": False, "message": "error in sign out"})
+
+
+# function to change password of the user with put method
+# We will expect 3 string values token, oldPassword, newPassword: .
+# if the data is valid we will change password of the user
+@app.route("/change_password", methods=["PUT"])
+def change_password():
+    token = request.headers.get("token", "")
+    oldPassword = request.form.get("oldPassword", "")
+    newPassword = request.form.get("newPassword", "")
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."})
+
+    if len(oldPassword) == 0 or len(newPassword) == 0:
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        )
+
+    if len(oldPassword) < 8 or len(newPassword) < 8:
+        return json.dumps(
+            {
+                "success": False,
+                "message": "old password and new password length should be grater then 8.",
+            }
+        )
+
+    if oldPassword == newPassword:
+        return json.dumps(
+            {
+                "success": False,
+                "message": "old password and new password should be different.",
+            }
+        )
+
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."})
+
+    email = result["data"][0][1]
+
+    # find_user_by_email_password
+    result = database_helper.find_user_by_email_password(email, oldPassword)
+
+    if result["status"] == True and (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "Wrong password."})
+
+    result = database_helper.change_password(email, newPassword)
+    if result["status"] == True:
+        return json.dumps({"success": False, "message": "Password changed."})
+
+    return json.dumps({"success": False, "message": "error in change password"})
+
+
+# function to get the user with get method.
+# We will expect 2 string values email and token: .
+# if the data is valid we return the user data
+@app.route("/get_user_data_by_email", methods=["GET"])
+def get_user_data_by_email():
+    token = request.headers.get("token", "")
+    email = request.form.get("email", "")
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."})
+
+    if len(email) == 0:
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        )
+
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."})
+
+    # find_user_by_email_password
+    result = database_helper.find_user_by_email(email)
+
+    if result["status"] == True and (len(result["data"])) != 0:
+        temp_user = {
+            "email": result["data"][0][0],
+            "firstname": result["data"][0][2],
+            "familyname": result["data"][0][3],
+            "gender": result["data"][0][4],
+            "city": result["data"][0][5],
+            "country": result["data"][0][6],
+        }
+        return json.dumps(
+            {"success": True, "message": "user details", "data": temp_user}
+        )
+    elif result["status"] == True and (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "Invalid user email"})
+
+    return json.dumps({"success": False, "message": "error in get_user_data_by_email"})
+
+
+# function to get the user with post method
+# We will expect 1 string values token: .
+# if the data is valid we return the user data
+@app.route("/get_user_data_by_token", methods=["GET"])
+def get_user_data_by_token():
+    token = request.headers.get("token", "")
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."})
+
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."})
+
+    email = result["data"][0][1]
+
+    # find_user_by_email_password
+    result = database_helper.find_user_by_email(email)
+
+    if result["status"] == True and (len(result["data"])) != 0:
+        temp_user = {
+            "email": result["data"][0][0],
+            "firstname": result["data"][0][2],
+            "familyname": result["data"][0][3],
+            "gender": result["data"][0][4],
+            "city": result["data"][0][5],
+            "country": result["data"][0][6],
+        }
+        return json.dumps(
+            {"success": True, "message": "user details", "data": temp_user}
+        )
+
+    return json.dumps({"success": False, "message": "error in get_user_data_by_token"})
+
+
+# function to get the user messages post method
+# We will expect 1 string values token: .
+# if the data is valid we return the user data
+@app.route("/get_user_messages_by_token", methods=["GET"])
+def get_user_messages_by_token():
+    token = request.headers.get("token", "")
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."})
+
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."})
+
+    email = result["data"][0][1]
+
+    # find_user_by_email_password
+    result = database_helper.get_message_from_email(email)
+
+    if result["status"] == True and (len(result["data"])) == 0:
+        return json.dumps({"success": True, "message": "No messages available"})
+    elif result["status"] == True and (len(result["data"])) != 0:
+        temp_data = []
+        for message in result["data"]:
+            temp_data.append({"email": message[2], "message": message[1]})
+        return json.dumps({"success": True, "data": temp_data})
+
+    return json.dumps({"success": False, "message": "error in get user data by token"})
+
+
+# function to get the user messages post method
+# We will expect 1 string values email: .
+# if the data is valid we return the user data
+@app.route("/get_user_messages_by_email", methods=["GET"])
+def get_user_messages_by_email():
+    token = request.headers.get("token", "")
+    email = request.form.get("email", "")
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."})
+
+    if len(email) == 0:
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        )
+
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."})
+
+    # validate email
+    result = database_helper.find_user_by_email(email)
+    if result["status"] == True and (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "Invalid user email"})
+
+    # find_user_by_email_password
+    result = database_helper.get_message_from_email(email)
+
+    if result["status"] == True and (len(result["data"])) == 0:
+        return json.dumps({"success": True, "message": "No messages available"})
+    elif result["status"] == True and (len(result["data"])) != 0:
+        temp_data = []
+        for message in result["data"]:
+            temp_data.append({"email": message[2], "message": message[1]})
+        return json.dumps({"success": True, "data": temp_data})
+
+    return json.dumps({"success": False, "message": "error in get user data by token"})
+
+
+# function to get the post messages
+# We will expect 3 string values token, message, email.
+# if the data is valid we return the user data
+@app.route("/post_message", methods=["POST"])
+def post_message():
+    recipient_email = request.form.get("email", "")
+    token = request.headers.get("token", "")
+    message = request.form.get("message", "")
+
+    if len(token) == 0:
+        return json.dumps({"success": False, "message": "Token missing or incorrect."})
+
+    if len(recipient_email) == 0 or len(message) == 0:
+        return json.dumps(
+            {"success": False, "message": "Form data missing or incorrect type."}
+        )
+
+    # validate token
+    result = database_helper.find_token(token)
+    if (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "You are not logged in."})
+    curr_user_email = result["data"][0][1]
+
+    # validate email
+    result = database_helper.find_user_by_email(recipient_email)
+    if result["status"] == True and (len(result["data"])) == 0:
+        return json.dumps({"success": False, "message": "Invalid user email"})
+
+    result = database_helper.add_message_and_email(
+        curr_user_email, message, recipient_email
+    )
+
+    if result["status"] == True:
+        return json.dumps({"success": True, "message": "Message posted"})
+
+    return json.dumps({"success": False, "message": "error in get_user_data_by_token"})
+
+
+app.run(host="0.0.0.0", port=5000)