jwt middleware integration

Author: MaliDipak

config/config.py

@@ -0,0 +1,7 @@
+dbconfig = {
+    "hostname": "localhost",
+    "username": "root",
+    "password": "radheradhe",
+    "dbname": "tset2",
+}
+

controller/userController.py

@@ -1,14 +1,19 @@
 from app import app
 from model.userModel import UserModel
+from middleware.jwt_auth import JWT_Auth
 from flask import request, make_response, jsonify
+from datetime import datetime
 
 obj = UserModel()
+auth = JWT_Auth()
 
 @app.route("/test")
 def test():
     return "test"
 
+
 @app.route("/user/getall")
+@auth.token_auth()
 def get_all_route():
     try:
         response = obj.get_all()
@@ -17,6 +22,7 @@ def get_all_route():
         return make_response(jsonify({"error": str(e)}), 500)
 
 @app.route("/user/getone")
+@auth.token_auth()
 def get_one_route():
     try:
         response = obj.get_one()
@@ -77,3 +83,47 @@ def delete_all_users_route():
         return response
     except Exception as e:
         return make_response(jsonify({"error": str(e)}), 500)
+
+
+@app.route("/user/getall/limit/<limit>/page/<page>", methods=["GET"])
+def get_all_users_by_limit_route(limit,page):
+    try:
+        response = obj.get_all_user_pagination_model(limit,page)
+        return response
+    except Exception as e:
+        return make_response(jsonify({"error": str(e)}), 500)
+    
+
+@app.route("/user/<uid>/upload/avatar", methods=["PUT"])
+def user_avatar_route(uid):
+    file = request.files['avatar']
+    unique_name = str(datetime.now().timestamp()).replace(".","")
+    file_extension = file.filename.split(".").pop()
+    file_name = f"{unique_name}.{file_extension}"
+    file.save(f"data/{file_name}")
+    try:
+        response = obj.avatar_model(file_name,uid)
+        return response
+    except Exception as e:
+        return  make_response(jsonify({"error": str(e)}), 500)
+
+
+@app.route("/user/avatar/<filename>",methods=["GET"])
+def user_get_avatar_route(filename):
+    try:
+        response = obj.get_avatar_model(filename)
+    except Exception as e:
+        return  make_response(jsonify({"error": str(e)}), 500)
+    else:
+        return response
+    
+
+@app.route("/user/login", methods=["POST"])
+def user_login_route():
+    data = request.form
+    try:
+        response = obj.user_login_model(data)
+    except Exception as e:
+        return  make_response(jsonify({"error": str(e)}), 500)
+    else:
+        return response

data/1708277547198316.jpg

@@ -0,0 +1,75 @@
+import mysql.connector
+import json
+import random
+from flask import make_response, jsonify, send_file, request
+from datetime import datetime, timedelta
+import jwt
+import re
+import ast
+from functools import wraps
+
+class JWT_Auth:
+    def __init__(self):
+        try:
+            self.con = mysql.connector.connect(host="localhost", user="root", password="radheradhe", database="tset2")
+            self.cur = self.con.cursor(dictionary=True)
+            print('Connection successful')
+        except Exception as e:
+            print("Some error: ", e)
+
+
+    def token_auth(self,endpoint=""):
+        def inner1(func):
+
+            @wraps(func)
+
+            def inner2(*args):
+
+                endpoint = request.url_rule
+                
+
+                Authorization_token = request.headers.get("Authorization")
+                
+                if re.match("^Bearer *([^ ]+) *$", Authorization_token, flags=0):
+                    token = Authorization_token.split(" ").pop()
+
+                    try:
+                        auth_user_data = jwt.decode(token, "hello", algorithms="HS256")
+                    except jwt.ExpiredSignatureError:
+                        return make_response(make_response({"ERROR":"TOKEN_EXPIRED"}, 401))
+                    
+                    
+                    role_id = auth_user_data["payload"]["role_id"]
+
+                    self.cur.execute(f"select roles from accessibility_view where endpoint = '{endpoint}'")
+                   
+                    res = self.cur.fetchall()
+                    
+                    if len(res)>0:
+
+                        roles = res[0]["roles"]
+
+                        # roles = ast.literal_eval(roles)
+
+
+                        if role_id in json.loads(roles):
+
+                            return func(*args)
+                        else:
+                            return make_response({"ERROR":"INVALID_ROLE"}, 401)
+        
+            
+                    
+                    else:
+                        return make_response({"ERROR":"end point not found"}, 401)
+        
+
+                    
+                else:
+                    return make_response({"ERROR":"Invalid Token"}, 401)
+            
+            return inner2
+        
+        return inner1
+        
+ 

data/1708277549119973.jpg

@@ -1,12 +1,15 @@
 import mysql.connector
 import json
 import random
-from flask import make_response, jsonify
+from flask import make_response, jsonify, send_file
+from datetime import datetime, timedelta
+import jwt
+from config.config import dbconfig
 
 class UserModel:
     def __init__(self):
         try:
-            self.con = mysql.connector.connect(host="localhost", user="root", password="radheradhe", database="tset2")
+            self.con = mysql.connector.connect(host=dbconfig["hostname"], user=dbconfig["username"], password=dbconfig["password"], database=dbconfig["dbname"])
             self.cur = self.con.cursor(dictionary=True)
             print('Connection successful')
         except Exception as e:
@@ -128,3 +131,70 @@ def delete_all_users(self):
                 return make_response(jsonify({"message": "All users deleted"}), 200)
             else:
                 return make_response(jsonify({"message": "Either users not found or no data to change"}), 204)
+
+    
+    def get_all_user_pagination_model(self,limit,page):
+        try:
+            limit = int(limit)
+            page = int(page)
+            start = (page*limit)-limit
+            query = f"SELECT * FROM new_table limit {start},{limit}"
+            self.cur.execute(query)
+            data = self.cur.fetchall()
+            if data:
+                res = make_response(jsonify({"payload": data}), 200)
+                # res.headers["Access-Control-Allow-Origin"] = "*"  # CORS  now there is no need to implement it here for all the methods.
+                #  it is implemented in app.py file 
+                return res
+            else:
+                return make_response(jsonify({"message": "No data found"}), 204)
+        except Exception as e:
+            return make_response(jsonify({"error": f"Error fetching data: {str(e)}"}), 500)
+
+
+    def avatar_model(self, file_name,uid):
+        try:
+            query = f"UPDATE new_table SET avatar='{file_name}' WHERE id={uid}"
+            self.cur.execute(query)
+            row_affected = self.cur.rowcount
+        except Exception as e:
+            return make_response(jsonify({"error": str(e)}), 500)
+        else:
+            self.con.commit()
+            if row_affected > 0:
+                return make_response(jsonify({"message": f"User {id} updated"}), 200)
+            else:
+                return make_response(jsonify({"message": f"User {id} not found or something went wrong!"}), 204)
+            
+    def get_avatar_model(self, file_name):
+        try:
+            return send_file(f"data/{file_name}")
+        except Exception as e:
+            return make_response(jsonify({"message": f"User {id} not found or something went wrong!"}), 204)
+
+    
+    def user_login_model(self,data):
+        id = data["id"]
+        # name = data["name"]
+        try:
+            query = f"SELECT * FROM new_table WHERE id={id}"
+            self.cur.execute(query)
+            data = self.cur.fetchall()
+            user_data = data[0]
+            exp_time = datetime.now() + timedelta(minutes=15)
+            exp_epoch_time = int(exp_time.timestamp())
+            secrete_key = "hello"
+            payload = {
+                "payload":user_data,
+                "exp":exp_epoch_time
+            }
+            token = jwt.encode(payload, secrete_key, algorithm="HS256")
+
+            if data:
+                res = make_response({"token":token}, 200)
+                return res
+            else:
+                return make_response(jsonify({"message": "No data found"}), 204)
+        except Exception as e:
+            return make_response(jsonify({"error": f"Error fetching data: {str(e)}"}), 500)
+