user module implementation

.gitignore

@@ -150,3 +150,4 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+*.db

.pylintrc

@@ -0,0 +1 @@
+ignored-classes=SQLObject,Registrant,scoped_session

Makefile

@@ -5,7 +5,7 @@ install:
 	pip install -r requirements.txt
 
 run:
-	flask --app main run 
+	flask --app main run --host=0.0.0.0
 
 dev:
-	flask --app main --debug run 
+	flask --app main --debug run

blueprints/auth.py

@@ -0,0 +1,63 @@
+'''
+    auth blueprint
+'''
+
+
+from flask import Blueprint, request
+from marshmallow import Schema, ValidationError, fields, validate
+from services.jwt_wrapper.jwt_wrapper import JwtWrapper
+from services.user.user import UserService
+from util.validate_request import ValidateRequest
+
+bp = Blueprint("auth", __name__, url_prefix="/auth")
+
+
+class LoginDto(Schema):
+    email = fields.Email(required=True)
+    password = fields.Str(
+        required=True,
+        validate=validate.Length(min=8, max=50)
+    )
+
+
+class RegisterDto(LoginDto):
+    pass
+
+
+@bp.route('/login', methods=['POST'])
+def login():
+    data = ValidateRequest(LoginDto, request)
+    check = UserService.login(data.email, data.password)
+
+    if check:
+        user = UserService.find_by_email(data.email)
+        jwt = {"user_id": user.id}
+        token = JwtWrapper.encode(jwt, 60*5)
+        return {'token': token}, 200
+    return {"message": "Invalid login!"}, 401
+
+
+@bp.route('/register', methods=['POST'])
+def register():
+    data = ValidateRequest(LoginDto, request)
+    UserService.register(data.email, data.password)
+    return {'message': "Successfully registered"}, 201
+
+
+@bp.route('/user', methods=['GET'])
+def user():
+    jwt_token = request.headers.get("Authorization").split("Bearer ")[1]
+    check = JwtWrapper.validate(jwt_token)
+
+    if check:
+        decoded = JwtWrapper.decode(jwt_token)
+        user = UserService.find_by_id(decoded["user_id"])
+        return {"id": user.id, "email": user.email}, 200
+    return {"message": "Unauthorized!"}, 401
+
+
+@bp.errorhandler(Exception)
+def error_handle(err: Exception):
+    if err.__class__ is ValidationError:
+        return str(err), 400
+    return {'message': "Internal server exception!", "error": str(err)}, 500

main.py

@@ -1,8 +1,26 @@
+'''
+    main
+'''
+
 from flask import Flask
 
+import blueprints.auth as auth
+from services.database import db_session, init_db
+
 app = Flask(__name__)
 
 
+app.register_blueprint(auth.bp)
+
+
+@app.teardown_appcontext
+def shutdown_session(__exception=None):
+    db_session.remove()
+
+
+init_db()
+
+
 @app.route("/")
 def hello_world():
     return "<p>Hello, World!</p>"

requirements.txt

@@ -1,20 +1,65 @@
-astroid==2.12.10
-autopep8==1.7.0
-click==8.1.3
-colorama==0.4.5
-dill==0.3.5.1
-Flask==2.2.2
+anyio==3.6.1
+asgiref==3.5.2
+astroid==2.9.3
+autopep8==1.6.0
+black==21.12b0
+certifi==2022.6.15
+charset-normalizer==2.1.0
+click==8.0.3
+colorama==0.4.4
+decorator==4.4.2
+fastapi==0.78.0
+flake8==4.0.1
+Flask==2.0.3
+greenlet==1.1.2
+h11==0.13.0
+httptools==0.4.0
+idna==3.3
+imageio==2.19.3
+imageio-ffmpeg==0.4.7
 isort==5.10.1
-itsdangerous==2.1.2
-Jinja2==3.1.2
+itsdangerous==2.1.0
+Jinja2==3.0.3
 lazy-object-proxy==1.7.1
-MarkupSafe==2.1.1
-mccabe==0.7.0
-platformdirs==2.5.2
-pycodestyle==2.9.1
-pylint==2.15.2
+MarkupSafe==2.1.0
+mccabe==0.6.1
+MouseInfo==0.1.3
+moviepy==1.0.3
+mypy-extensions==0.4.3
+numpy==1.23.1
+pathspec==0.9.0
+Pillow==9.1.1
+platformdirs==2.4.1
+proglog==0.1.10
+PyAutoGUI==0.9.53
+pycodestyle==2.8.0
+pydantic==1.9.1
+PyDirectInput==1.0.4
+pyflakes==2.4.0
+PyGetWindow==0.0.9
+pylint==2.12.2
+PyMsgBox==1.0.9
+pynput==1.7.6
+pyperclip==1.8.2
+PyRect==0.2.0
+PyScreeze==0.1.28
+pyserial==3.5
+python-dotenv==0.20.0
+pytweening==1.0.4
+pywin32==304
+PyYAML==6.0
+requests==2.28.1
+six==1.16.0
+sniffio==1.2.0
+SQLAlchemy==1.4.31
+starlette==0.19.1
 toml==0.10.2
-tomli==2.0.1
-tomlkit==0.11.4
-Werkzeug==2.2.2
-wrapt==1.14.1
+tomli==1.2.3
+tqdm==4.64.0
+typing_extensions==4.0.1
+urllib3==1.26.10
+uvicorn==0.17.6
+watchgod==0.8.2
+websockets==10.3
+Werkzeug==2.0.3
+wrapt==1.13.3

services/bcrypt_wrapper/bcrypt_wrapper.py

@@ -0,0 +1,21 @@
+"""
+    bcrypt wrapper module
+"""
+
+
+import os
+
+import bcrypt
+
+
+class BcryptWrapper:
+    __salt_round = os.getenv("BCRYPT_SALT") or 10
+    __salt = bcrypt.gensalt(rounds=__salt_round)
+
+    @staticmethod
+    def hash(value: str):
+        return bcrypt.hashpw(value.encode("utf-8"), BcryptWrapper.__salt)
+
+    @staticmethod
+    def validate(value: str, hash: str):
+        return bcrypt.checkpw(value.encode("utf-8"), hash)

services/database.py

@@ -0,0 +1,21 @@
+"""
+    database module
+"""
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import scoped_session, sessionmaker
+
+engine = create_engine('sqlite:///test.db')
+db_session = scoped_session(
+    sessionmaker(autocommit=False, autoflush=False, bind=engine)
+)
+
+Base = declarative_base()
+Base.query = db_session.query_property()
+
+
+def init_db():
+    # import all modules here that might define models so that
+    # they will be registered properly on the metadata.  Otherwise
+    # you will have to import them first before calling init_db()
+    Base.metadata.create_all(bind=engine)

services/jwt_wrapper/jwt_wrapper.py

@@ -0,0 +1,28 @@
+"""
+    jwt wrapper module
+"""
+
+
+import os
+from datetime import datetime, timedelta, timezone
+
+import jwt
+
+
+class JwtWrapper:
+    __jwt_secret = os.getenv("BCRYPT_SALT") or "dev-secret"
+
+    def encode(value: any, duration: int):
+        value["iat"] = datetime.now(tz=timezone.utc)
+        value["exp"] = value["iat"] + timedelta(seconds=duration)
+        return jwt.encode(value, JwtWrapper.__jwt_secret, "HS256")
+
+    def validate(value: str):
+        try:
+            jwt.decode(value, JwtWrapper.__jwt_secret, algorithms=["HS256"])
+            return True
+        except:
+            return False
+
+    def decode(value: str):
+        return jwt.decode(value, JwtWrapper.__jwt_secret, algorithms=["HS256"])

services/user/user.py

@@ -0,0 +1,47 @@
+from services.bcrypt_wrapper.bcrypt_wrapper import BcryptWrapper
+from services.database import Base, db_session
+from sqlalchemy import Column, Integer, String
+from sqlalchemy.exc import IntegrityError
+
+
+class User(Base):
+    __tablename__ = 'users'
+
+    id = Column(Integer, primary_key=True)
+    email = Column(String(100), unique=True)
+    password = Column(String(100))
+
+    def __init__(self, email: str, password: str):
+        self.email = email
+        self.password = password
+
+    def __repr__(self):
+        return f'<User {self.email!r}>'
+
+
+class UserService:
+    @staticmethod
+    def register(email: str, password: str):
+        try:
+            hashed_password = BcryptWrapper.hash(password)
+            user = User(email, hashed_password)
+            db_session.add(user)
+            db_session.commit()
+        except IntegrityError:
+            raise Exception("User already registerd!")
+
+    @staticmethod
+    def login(email: str, password: str):
+        user: User = UserService.find_by_email(email)
+        check_pwd = BcryptWrapper.validate(password, user.password)
+        return check_pwd
+
+    @staticmethod
+    def find_by_email(email: str):
+        user: User = User.query.filter(User.email == email).first()
+        return user
+
+    @staticmethod
+    def find_by_id(id: int):
+        user: User = User.query.filter(User.id == id).first()
+        return user

util/validate_request.py

@@ -0,0 +1,20 @@
+"""
+    function for validate request json
+"""
+
+from collections import namedtuple
+from typing import TypeVar
+
+from flask import Request
+from marshmallow import ValidationError
+
+T = TypeVar('T')
+
+
+def ValidateRequest(schema: T, request: Request) -> T:
+    template = schema()
+    dict = request.get_json()
+    err = template.validate(dict)
+    if err:
+        raise ValidationError(err)
+    return namedtuple(schema.__class__.__name__, dict.keys())(*dict.values())