Upload 9 files

.env.sample

@@ -0,0 +1,2 @@
+DATABASE_URL=
+SECRET_KEY=

.gitignore

@@ -0,0 +1,5 @@
+venv
+env
+.env
+*.pyc
+static/images/

README.md

@@ -1,10 +1,9 @@
----
-title: TestingSpace
-emoji: ⚡
-colorFrom: pink
-colorTo: purple
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Authentication with Flask and JWT
+
+## Development server
+
+- Create a virtual environment, activate it and install the dependencies.
+- Run `flask run` to start the development server.
+- Navigate to http://localhost:5000/. The app will automatically reload if you change any of the source files.
+
+© 2021 GitHub, Inc.

app.py

@@ -0,0 +1,295 @@
+import jwt, os
+from dotenv import load_dotenv
+from flask import Flask, request, jsonify
+from save_image import save_pic
+from validate import validate_book, validate_email_and_password, validate_user
+
+load_dotenv()
+
+app = Flask(__name__)
+SECRET_KEY = os.environ.get('SECRET_KEY') or 'myFlask@uthApp!s$uper8'
+print(SECRET_KEY)
+app.config['SECRET_KEY'] = SECRET_KEY
+
+from models import Books, User
+from auth_middleware import token_required
+
+@app.route("/")
+def hello():
+    return "Hello World!"
+
+@app.route("/users/", methods=["POST"])
+def add_user():
+    try:
+        user = request.json
+        if not user:
+            return {
+                "message": "Please provide user details",
+                "data": None,
+                "error": "Bad request"
+            }, 400
+        is_validated = validate_user(**user)
+        if is_validated is not True:
+            return dict(message='Invalid data', data=None, error=is_validated), 400
+        user = User().create(**user)
+        if not user:
+            return {
+                "message": "User already exists",
+                "error": "Conflict",
+                "data": None
+            }, 409
+        return {
+            "message": "Successfully created new user",
+            "data": user
+        }, 201
+    except Exception as e:
+        return {
+            "message": "Something went wrong",
+            "error": str(e),
+            "data": None
+        }, 500
+
+@app.route("/users/login", methods=["POST"])
+def login():
+    try:
+        data = request.json
+        if not data:
+            return {
+                "message": "Please provide user details",
+                "data": None,
+                "error": "Bad request"
+            }, 400
+        # validate input
+        is_validated = validate_email_and_password(data.get('email'), data.get('password'))
+        if is_validated is not True:
+            return dict(message='Invalid data', data=None, error=is_validated), 400
+        user = User().login(
+            data["email"],
+            data["password"]
+        )
+        if user:
+            try:
+                # token should expire after 24 hrs
+                user["token"] = jwt.encode(
+                    {"user_id": user["_id"]},
+                    app.config["SECRET_KEY"],
+                    algorithm="HS256"
+                )
+                return {
+                    "message": "Successfully fetched auth token",
+                    "data": user
+                }
+            except Exception as e:
+                return {
+                    "error": "Something went wrong",
+                    "message": str(e)
+                }, 500
+        return {
+            "message": "Error fetching auth token!, invalid email or password",
+            "data": None,
+            "error": "Unauthorized"
+        }, 404
+    except Exception as e:
+        return {
+                "message": "Something went wrong!",
+                "error": str(e),
+                "data": None
+        }, 500
+
+
+@app.route("/users/", methods=["GET"])
+@token_required
+def get_current_user(current_user):
+    return jsonify({
+        "message": "successfully retrieved user profile",
+        "data": current_user
+    })
+
+@app.route("/users/", methods=["PUT"])
+@token_required
+def update_user(current_user):
+    try:
+        user = request.json
+        if user.get("name"):
+            user = User().update(current_user["_id"], user["name"])
+            return jsonify({
+                "message": "successfully updated account",
+                "data": user
+            }), 201
+        return {
+            "message": "Invalid data, you can only update your account name!",
+            "data": None,
+            "error": "Bad Request"
+        }, 400
+    except Exception as e:
+        return jsonify({
+            "message": "failed to update account",
+            "error": str(e),
+            "data": None
+        }), 400
+
+@app.route("/users/", methods=["DELETE"])
+@token_required
+def disable_user(current_user):
+    try:
+        User().disable_account(current_user["_id"])
+        return jsonify({
+            "message": "successfully disabled acount",
+            "data": None
+        }), 204
+    except Exception as e:
+        return jsonify({
+            "message": "failed to disable account",
+            "error": str(e),
+            "data": None
+        }), 400
+
+@app.route("/books/", methods=["POST"])
+@token_required
+def add_book(current_user):
+    try:
+        book = dict(request.form)
+        if not book:
+            return {
+                "message": "Invalid data, you need to give the book title, cover image, author id,",
+                "data": None,
+                "error": "Bad Request"
+            }, 400
+        if not request.files["cover_image"]:
+            return {
+                "message": "cover image is required",
+                "data": None
+            }, 400
+
+        book["image_url"] = request.host_url+"static/books/"+save_pic(request.files["cover_image"])
+        is_validated = validate_book(**book)
+        if is_validated is not True:
+            return {
+                "message": "Invalid data",
+                "data": None,
+                "error": is_validated
+            }, 400
+        book = Books().create(**book, user_id=current_user["_id"])
+        if not book:
+            return {
+                "message": "The book has been created by user",
+                "data": None,
+                "error": "Conflict"
+            }, 400
+        return jsonify({
+            "message": "successfully created a new book",
+            "data": book
+        }), 201
+    except Exception as e:
+        return jsonify({
+            "message": "failed to create a new book",
+            "error": str(e),
+            "data": None
+        }), 500
+
+@app.route("/books/", methods=["GET"])
+@token_required
+def get_books(current_user):
+    try:
+        books = Books().get_by_user_id(current_user["_id"])
+        return jsonify({
+            "message": "successfully retrieved all books",
+            "data": books
+        })
+    except Exception as e:
+        return jsonify({
+            "message": "failed to retrieve all books",
+            "error": str(e),
+            "data": None
+        }), 500
+
+@app.route("/books/<book_id>", methods=["GET"])
+@token_required
+def get_book(book_id):
+    try:
+        book = Books().get_by_id(book_id)
+        if not book:
+            return {
+                "message": "Book not found",
+                "data": None,
+                "error": "Not Found"
+            }, 404
+        return jsonify({
+            "message": "successfully retrieved a book",
+            "data": book
+        })
+    except Exception as e:
+        return jsonify({
+            "message": "Something went wrong",
+            "error": str(e),
+            "data": None
+        }), 500
+
+@app.route("/books/<book_id>", methods=["PUT"])
+@token_required
+def update_book(current_user, book_id):
+    try:
+        book = Books().get_by_id(book_id)
+        if not book or book["user_id"] != current_user["_id"]:
+            return {
+                "message": "Book not found for user",
+                "data": None,
+                "error": "Not found"
+            }, 404
+        book = request.form
+        if book.get('cover_image'):
+            book["image_url"] = request.host_url+"static/books/"+save_pic(request.files["cover_image"])
+        book = Books().update(book_id, **book)
+        return jsonify({
+            "message": "successfully updated a book",
+            "data": book
+        }), 201
+    except Exception as e:
+        return jsonify({
+            "message": "failed to update a book",
+            "error": str(e),
+            "data": None
+        }), 400
+
+@app.route("/books/<book_id>", methods=["DELETE"])
+@token_required
+def delete_book(current_user, book_id):
+    try:
+        book = Books().get_by_id(book_id)
+        if not book or book["user_id"] != current_user["_id"]:
+            return {
+                "message": "Book not found for user",
+                "data": None,
+                "error": "Not found"
+            }, 404
+        Books().delete(book_id)
+        return jsonify({
+            "message": "successfully deleted a book",
+            "data": None
+        }), 204
+    except Exception as e:
+        return jsonify({
+            "message": "failed to delete a book",
+            "error": str(e),
+            "data": None
+        }), 400
+
+@app.errorhandler(403)
+def forbidden(e):
+    return jsonify({
+        "message": "Forbidden",
+        "error": str(e),
+        "data": None
+    }), 403
+
+@app.errorhandler(404)
+def forbidden(e):
+    return jsonify({
+        "message": "Endpoint Not Found",
+        "error": str(e),
+        "data": None
+    }), 404
+
+
+if __name__ == "__main__":
+    app.run(debug=True)

auth_middleware.py

@@ -0,0 +1,39 @@
+from functools import wraps
+import jwt
+from flask import request, abort
+from flask import current_app
+import models
+
+def token_required(f):
+    @wraps(f)
+    def decorated(*args, **kwargs):
+        token = None
+        if "Authorization" in request.headers:
+            token = request.headers["Authorization"].split(" ")[1]
+        if not token:
+            return {
+                "message": "Authentication Token is missing!",
+                "data": None,
+                "error": "Unauthorized"
+            }, 401
+        try:
+            data=jwt.decode(token, current_app.config["SECRET_KEY"], algorithms=["HS256"])
+            current_user=models.User().get_by_id(data["user_id"])
+            if current_user is None:
+                return {
+                "message": "Invalid Authentication token!",
+                "data": None,
+                "error": "Unauthorized"
+            }, 401
+            if not current_user["active"]:
+                abort(403)
+        except Exception as e:
+            return {
+                "message": "Something went wrong",
+                "data": None,
+                "error": str(e)
+            }, 500
+
+        return f(current_user, *args, **kwargs)
+
+    return decorated

models.py

@@ -0,0 +1,181 @@
+"""Application Models"""
+import bson, os
+from dotenv import load_dotenv
+from pymongo import MongoClient
+from werkzeug.security import generate_password_hash, check_password_hash
+
+load_dotenv()
+
+DATABASE_URL=os.environ.get('DATABASE_URL') or 'mongodb://localhost:27017/flask-auth'
+print(DATABASE_URL)
+client = MongoClient(DATABASE_URL)
+db = client.myDatabase
+
+class Books:
+    """Books Model"""
+    def __init__(self):
+        return
+
+    def create(self, title="", description="", image_url="", category="", user_id=""):
+        """Create a new book"""
+        book = self.get_by_user_id_and_title(user_id, title)
+        if book:
+            return
+        new_book = db.books.insert_one(
+            {
+                "title": title,
+                "description": description,
+                "image_url": image_url,
+                "category": category,
+                "user_id": user_id
+            }
+        )
+        return self.get_by_id(new_book.inserted_id)
+
+    def get_all(self):
+        """Get all books"""
+        books = db.books.find()
+        return [{**book, "_id": str(book["_id"])} for book in books]
+
+    def get_by_id(self, book_id):
+        """Get a book by id"""
+        book = db.books.find_one({"_id": bson.ObjectId(book_id)})
+        if not book:
+            return
+        book["_id"] = str(book["_id"])
+        return book
+
+    def get_by_user_id(self, user_id):
+        """Get all books created by a user"""
+        books = db.books.find({"user_id": user_id})
+        return [{**book, "_id": str(book["_id"])} for book in books]
+       
+    def get_by_category(self, category):
+        """Get all books by category"""
+        books = db.books.find({"category": category})
+        return [book for book in books]
+
+    def get_by_user_id_and_category(self, user_id, category):
+        """Get all books by category for a particular user"""
+        books = db.books.find({"user_id": user_id, "category": category})
+        return [{**book, "_id": str(book["_id"])} for book in books]
+
+    def get_by_user_id_and_title(self, user_id, title):
+        """Get a book given its title and author"""
+        book = db.books.find_one({"user_id": user_id, "title": title})
+        if not book:
+            return
+        book["_id"] = str(book["_id"])
+        return book
+
+    def update(self, book_id, title="", description="", image_url="", category="", user_id=""):
+        """Update a book"""
+        data={}
+        if title: data["title"]=title
+        if description: data["description"]=description
+        if image_url: data["image_url"]=image_url
+        if category: data["category"]=category
+
+        book = db.books.update_one(
+            {"_id": bson.ObjectId(book_id)},
+            {
+                "$set": data
+            }
+        )
+        book = self.get_by_id(book_id)
+        return book
+
+    def delete(self, book_id):
+        """Delete a book"""
+        book = db.books.delete_one({"_id": bson.ObjectId(book_id)})
+        return book
+
+    def delete_by_user_id(self, user_id):
+        """Delete all books created by a user"""
+        book = db.books.delete_many({"user_id": bson.ObjectId(user_id)})
+        return book
+
+
+class User:
+    """User Model"""
+    def __init__(self):
+        return
+
+    def create(self, name="", email="", password=""):
+        """Create a new user"""
+        user = self.get_by_email(email)
+        if user:
+            return
+        new_user = db.users.insert_one(
+            {
+                "name": name,
+                "email": email,
+                "password": self.encrypt_password(password),
+                "active": True
+            }
+        )
+        return self.get_by_id(new_user.inserted_id)
+
+    def get_all(self):
+        """Get all users"""
+        users = db.users.find({"active": True})
+        return [{**user, "_id": str(user["_id"])} for user in users]
+
+    def get_by_id(self, user_id):
+        """Get a user by id"""
+        user = db.users.find_one({"_id": bson.ObjectId(user_id), "active": True})
+        if not user:
+            return
+        user["_id"] = str(user["_id"])
+        user.pop("password")
+        return user
+
+    def get_by_email(self, email):
+        """Get a user by email"""
+        user = db.users.find_one({"email": email, "active": True})
+        if not user:
+            return
+        user["_id"] = str(user["_id"])
+        return user
+
+    def update(self, user_id, name=""):
+        """Update a user"""
+        data = {}
+        if name:
+            data["name"] = name
+        user = db.users.update_one(
+            {"_id": bson.ObjectId(user_id)},
+            {
+                "$set": data
+            }
+        )
+        user = self.get_by_id(user_id)
+        return user
+
+    def delete(self, user_id):
+        """Delete a user"""
+        Books().delete_by_user_id(user_id)
+        user = db.users.delete_one({"_id": bson.ObjectId(user_id)})
+        user = self.get_by_id(user_id)
+        return user
+
+    def disable_account(self, user_id):
+        """Disable a user account"""
+        user = db.users.update_one(
+            {"_id": bson.ObjectId(user_id)},
+            {"$set": {"active": False}}
+        )
+        user = self.get_by_id(user_id)
+        return user
+
+    def encrypt_password(self, password):
+        """Encrypt password"""
+        return generate_password_hash(password)
+
+    def login(self, email, password):
+        """Login a user"""
+        user = self.get_by_email(email)
+        if not user or not check_password_hash(user["password"], password):
+            return
+        user.pop("password")
+        return user

requirements.txt

@@ -0,0 +1,12 @@
+click==8.0.1
+colorama==0.4.6
+Flask==2.0.1
+Flask-PyMongo==2.3.0
+itsdangerous==2.0.1
+Jinja2==3.0.1
+MarkupSafe==2.0.1
+pillow==10.3.0
+PyJWT==2.1.0
+pymongo==3.12.0
+python-dotenv==0.19.2
+Werkzeug==2.0.1

save_image.py

@@ -0,0 +1,22 @@
+import secrets
+import os
+from PIL import Image
+from flask import current_app as app
+
+def save_pic(picture):
+    """Saves an image to disk"""
+    file_name = secrets.token_hex(8) +os.path.splitext(picture.filename)[1]
+    if not os.path.isdir(os.path.join(app.root_path, 'static')):
+        os.mkdir(os.path.join(app.root_path,"static"))
+        os.mkdir(os.path.join(app.root_path,"static/images"))
+        os.mkdir(os.path.join(app.root_path,"static/images/books"))
+    if not os.path.isdir(os.path.join(app.root_path, 'static/images')):
+        os.mkdir(os.path.join(app.root_path,"static/images"))
+        os.mkdir(os.path.join(app.root_path,"static/images/books"))
+    if not os.path.isdir(os.path.join(app.root_path, 'static/images/books')):
+        os.mkdir(os.path.join(app.root_path,"static/images/books"))
+    file_path = os.path.join(app.root_path, "static/images/books", file_name)
+    picture = Image.open(picture)
+    picture.thumbnail((150, 150))
+    picture.save(file_path)
+    return file_name

validate.py

@@ -0,0 +1,102 @@
+"""Validator Module"""
+import re
+from bson.objectid import ObjectId
+
+
+def validate(data, regex):
+    """Custom Validator"""
+    return True if re.match(regex, data) else False
+
+
+def validate_password(password: str):
+    """Password Validator"""
+    reg = r"\b^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)(?=.*[@$!%*#?&])[A-Za-z\d@$!#%*?&]{8,20}$\b"
+    return validate(password, reg)
+
+
+def validate_email(email: str):
+    """Email Validator"""
+    regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
+    return validate(email, regex)
+
+
+def validate_book(**args):
+    """Book Validator"""
+    if not args.get('title') or not args.get('image_url') \
+            or not args.get('category') or not args.get('user_id'):
+        return {
+            'title': 'Title is required',
+            'image_url': 'Image URL is required',
+            'category': 'Category is required',
+            'user_id': 'User ID is required'
+        }
+    if args.get('category') not in ['romance', 'peotry', 'politics' 'picture book', 'science', 'fantasy', 'horror', 'thriller']:
+        return {
+            'status': 'error',
+            'message': 'Invalid category'
+        }
+    try:
+        ObjectId(args.get('user_id'))
+    except:
+        return {
+            'user_id': 'User ID must be valid'
+        }
+    if not isinstance(args.get('title'), str) or not isinstance(args.get('description'), str) \
+            or not isinstance(args.get('image_url'), str):
+        return {
+            'title': 'Title must be a string',
+            'description': 'Description must be a string',
+            'image_url': 'Image URL must be a string'
+        }
+    return True
+
+
+def validate_user(**args):
+    """User Validator"""
+    if not args.get('email') or not args.get('password') or not args.get('name'):
+        return {
+            'email': 'Email is required',
+            'password': 'Password is required',
+            'name': 'Name is required'
+        }
+    if not isinstance(args.get('name'), str) or \
+            not isinstance(args.get('email'), str) or not isinstance(args.get('password'), str):
+        return {
+            'email': 'Email must be a string',
+            'password': 'Password must be a string',
+            'name': 'Name must be a string'
+        }
+    if not validate_email(args.get('email')):
+        return {
+            'email': 'Email is invalid'
+        }
+    if not validate_password(args.get('password')):
+        return {
+            'password': 'Password is invalid, Should be atleast 8 characters with \
+                upper and lower case letters, numbers and special characters'
+        }
+    print(len(args['name'].split(' ')))
+    if not 2 <= len(args['name'].split(' ')) <= 30:
+        return {
+            'name': 'Name must be between 2 and 30 words'
+        }
+    return True
+
+
+def validate_email_and_password(email, password):
+    """Email and Password Validator"""
+    if not (email and password):
+        return {
+            'email': 'Email is required',
+            'password': 'Password is required'
+        }
+    if not validate_email(email):
+        return {
+            'email': 'Email is invalid'
+        }
+    if not validate_password(password):
+        return {
+            'password': 'Password is invalid, Should be atleast 8 characters with \
+                upper and lower case letters, numbers and special characters'
+        }
+    return True