Initial commit

.gitignore

@@ -0,0 +1 @@
+.env

.vscode/launch.json

@@ -0,0 +1,24 @@
+{
+	// Pro informace o možných atributech použijte technologii IntelliSense.
+	// Umístěním ukazatele myši zobrazíte popisy existujících atributů.
+	// Další informace najdete tady: https://go.microsoft.com/fwlink/?linkid=830387
+	"version": "0.2.0",
+	"configurations": [
+		{
+			"name": "Flask Shop",
+			"type": "debugpy",
+			"request": "launch",
+			"program": "${workspaceFolder}/main.py",
+			"args": [],
+			"cwd": "${workspaceFolder}",
+			"env": {
+				"FLASK_APP": "main.py",
+				"FLASK_ENV": "development"
+			},
+			"envFile": "${workspaceFolder}/.env",
+			"stopOnEntry": false,
+			"console": "internalConsole",
+			"autoReload": {"enable": true}
+		}
+	]
+}

.vscode/settings.json

@@ -0,0 +1,6 @@
+{
+	"cSpell.words": [
+		"dotenv",
+		"jsonify"
+	]
+}

app/__init__.py

@@ -0,0 +1,17 @@
+from flask import Flask
+from flask_jwt_extended import JWTManager
+
+def create_app():
+	app = Flask(__name__)
+	jwt = JWTManager(app)
+
+	from app.api import bp, bp_errors, bp_product, bp_user
+	app.register_blueprint(bp)
+	app.register_blueprint(bp_errors)
+	app.register_blueprint(bp_product)
+	app.register_blueprint(bp_user)
+
+	from app.config import FlaskTesting, FlaskProduction
+	app.config.from_object(FlaskTesting)
+
+	return app

app/api/__init__.py

@@ -0,0 +1,8 @@
+from flask import Blueprint
+
+bp_errors = Blueprint('errors', __name__)
+bp = Blueprint('api', __name__)
+bp_product = Blueprint('product', __name__, url_prefix="/product")
+bp_user = Blueprint('user', __name__, url_prefix="/user")
+
+from . import routes

app/api/routes/__init__.py

@@ -0,0 +1 @@
+from . import main_routes,error_routes, product_routes, user_routes

app/api/routes/error_routes.py

@@ -0,0 +1,29 @@
+from .. import bp_errors
+
+@bp_errors.app_errorhandler(400)
+def bad_request(e):
+	return {"Bad Request": "The request was incorrectly formatted, or contained invalid data"}, 400
+
+@bp_errors.app_errorhandler(401)
+def unauthorized(e):
+	return {"Unauthorized": "Failed to authorize the request"}, 401
+
+@bp_errors.app_errorhandler(403)
+def forbidden(e):
+	return {"Forbidden": "Forbidden from accessing this resource. Try logging in"}, 403
+
+@bp_errors.app_errorhandler(404)
+def not_found(e):
+	return {"Not Found": "The requested resource was not found"}, 404
+
+@bp_errors.app_errorhandler(405)
+def method_not_allowed(e):
+	return {"Method Not Allowed": "The method used is not allowed in current context"}, 405
+
+@bp_errors.app_errorhandler(500)
+def internal_error(e):
+	return {"Internal Server Error": "An error occured on he server"}, 500
+
+@bp_errors.app_errorhandler(501)
+def internal_error(e):
+	return {"Not Implemented": "This function has not been implemented yet. Check back soon!"}, 501

app/api/routes/main_routes.py

@@ -0,0 +1,7 @@
+from flask import jsonify, abort
+
+from .. import bp
+
+@bp.route('/')
+def hello():
+	return jsonify({'message': 'Hello, Flask!'})

app/api/routes/product_routes.py

@@ -0,0 +1,63 @@
+from flask import jsonify, abort, request
+
+from app.api import bp_product
+
+from app.services.product_service import ProductService
+
+@bp_product.route('/<int:id>', methods=['GET'])
+def all_product_info(id: int):
+	result = ProductService.get_all_info(id)
+
+	if result is not None:
+		return jsonify(result)
+	else:
+		abort(404)
+
+@bp_product.route('/<int:id>/name', methods=['GET'])
+def get_name(id: int):
+	result = ProductService.get_name(id)
+
+	if result is not None:
+		return jsonify({"name": result})
+	else:
+		return abort(404)
+
+@bp_product.route('/<int:id>/manufacturer', methods=['GET'])
+def get_manufacturer(id: int):
+	result = ProductService.get_manufacturer(id)
+
+	if result is not None:
+		return jsonify({"name": result})
+	else:
+		return abort(404)
+
+@bp_product.route('/<int:id>/price', methods=['GET'])
+def get_price(id: int):
+	result = ProductService.get_price(id)
+
+	if result is not None:
+		return jsonify({"price": result})
+	else:
+		return abort(404)
+
+@bp_product.route('/<int:id>/image', methods=['GET'])
+def get_image(id: int):
+	result = ProductService.get_image(id)
+
+	if result is not None:
+		return jsonify({"image": result})
+	else:
+		return abort(404)
+
+@bp_product.route('/<int:id>/image_name', methods=['GET'])
+def get_image_name(id: int):
+	result = ProductService.get_image_name(id)
+
+	if result is not None:
+		return jsonify({"image_name": result})
+	else:
+		return abort(404)
+
+@bp_product.route('/create', methods=['POST'])
+def create_product_listing():
+	return abort(501)

app/api/routes/user_routes.py

@@ -0,0 +1,79 @@
+from app.api import bp_user
+from flask_jwt_extended import jwt_required, get_jwt_identity, get_jwt
+from flask import request, abort, jsonify
+from datetime import timedelta
+
+from app.services.user_service import UserService
+from app.extensions import jwt_redis_blocklist
+
+@bp_user.route('/login', methods=['POST'])
+def login():
+	username = request.json.get('username')
+	password = request.json.get('password')
+
+	if username is None or password is None:
+		return abort(400)
+
+	result, status_code = UserService.login(username, password)
+
+	return jsonify(**result), status_code
+
+@bp_user.route('/logout', methods=['DELETE'])
+@jwt_required()
+def logout():
+	jti = get_jwt()["jti"]
+	jwt_redis_blocklist.set(jti, "", ex=timedelta(days=1))
+
+	return {"Success": "Successfully logged out"}, 200
+
+@bp_user.route('/create', methods=['POST'])
+def create_user():
+	username = request.json.get('username')
+	email = request.json.get('email')
+	password = request.json.get('password')
+
+	if username is None or email is None or password is None:
+		return abort(400)
+
+	result, status_code = UserService.create_user(username, email, password)
+
+	return jsonify(**result), status_code
+
+@bp_user.route('/update/email', methods=['POST'])
+@jwt_required()
+def update_email():
+	username = get_jwt_identity()
+	new_mail = request.json.get('new_email')
+
+	if new_mail is None:
+		return abort(400)
+
+	result, status_code = UserService.update_email(username, new_mail)
+
+	return jsonify(**result), status_code
+
+@bp_user.route('/update/username', methods=['POST'])
+@jwt_required()
+def update_username():
+	username = get_jwt_identity()
+	new_username = request.json.get('new_username')
+
+	if new_username is None:
+		return abort(400)
+
+	result, status_code = UserService.update_username(username, new_username)
+
+	return jsonify(**result), status_code
+
+@bp_user.route('/update/password', methods=['POST'])
+@jwt_required()
+def update_password():
+	username = get_jwt_identity()
+	new_password = request.json.get('new_password')
+
+	if new_password is None:
+		return abort(400)
+
+	result, status_code = UserService.update_password(username, new_password)
+
+	return jsonify(**result), status_code

app/config.py

@@ -0,0 +1,22 @@
+import os
+
+class MySqlConfig:
+	MYSQL_USER = os.environ.get('MYSQL_USER')
+	MYSQL_DATABASE = os.environ.get('MYSQL_DATABASE')
+	MYSQL_HOST = os.environ.get('MYSQL_HOST')
+	MYSQL_PORT = os.environ.get('MYSQL_PORT')
+	MYSQL_PASSWORD = os.environ.get('MYSQL_PASSWORD')
+
+class RedisConfig:
+	REDIS_HOST = os.environ.get('REDIS_HOST')
+	REDIS_PORT = os.environ.get('REDIS_PORT')
+
+class FlaskProduction:
+	DEBUG = False
+	JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY')
+	SERVER_NAME = os.environ.get('HOST') + ':' + os.environ.get('PORT')
+
+class FlaskTesting:
+	DEBUG = True
+	JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY')
+	SERVER_NAME = os.environ.get('HOST') + ':' + os.environ.get('PORT')

app/extensions.py

@@ -0,0 +1,22 @@
+import mysql.connector
+import redis
+import os
+
+from app.config import RedisConfig
+from app.config import MySqlConfig
+
+db_connection = mysql.connector.connect(
+	host=MySqlConfig.MYSQL_HOST,
+	user=MySqlConfig.MYSQL_USER,
+	password=MySqlConfig.MYSQL_PASSWORD,
+	database=MySqlConfig.MYSQL_DATABASE,
+)
+
+db_cursor = db_connection.cursor()
+
+jwt_redis_blocklist = redis.StrictRedis(
+	host=RedisConfig.REDIS_HOST,
+	port=RedisConfig.REDIS_PORT,
+	db=0,
+	decode_responses=True
+)

app/jwt_utils.py

@@ -0,0 +1,15 @@
+from app.extensions import jwt_redis_blocklist
+
+from flask_jwt_extended import create_access_token
+from flask_jwt_extended import get_jwt
+from flask_jwt_extended import jwt_required
+from flask_jwt_extended import JWTManager
+
+@jwt.token_in_blocklist_loader
+def check_if_token_is_revoked(jwt_header, jwt_payload: dict) -> bool:
+	jti = jwt_payload["jti"]
+	token_in_redis = jwt_redis_blocklist.get(jti)
+
+	print(token_in_redis)
+	
+	return token_in_redis is not None

app/services/product_service.py

@@ -0,0 +1,52 @@
+import base64
+
+from ..extensions import db_cursor as cursor
+
+class ProductService:
+
+	@staticmethod
+	def get_name(product_id: int):
+		cursor.execute(f"select name from product where product.product_id = {product_id}")
+		result = cursor.fetchone()
+		return result[0]
+
+	@staticmethod
+	def get_manufacturer(product_id: int):
+		cursor.execute(f"select manufacturer from product where product.product_id = {product_id}")
+		result = cursor.fetchone()
+		return result[0]
+
+	@staticmethod
+	def get_price(product_id: int):
+		cursor.execute(f"select price_pc from product where product.product_id = {product_id}")
+		result = cursor.fetchone()
+		return result[0]
+
+	@staticmethod
+	def get_image(product_id: int):
+		cursor.execute(f"select image from product where product.product_id = {product_id}")
+		result = cursor.fetchone()
+		return base64.b64encode(result[0]).decode('utf-8')
+
+	@staticmethod
+	def get_image_name(product_id: int):
+		cursor.execute(f"select image_name from product where product.product_id = {product_id}")
+		result = cursor.fetchone()
+		return result[0]
+
+	@staticmethod
+	def get_all_info(product_id: int):
+		cursor.execute(f"select name,manufacturer,price_pc,image_name,image from product where product.product_id = {product_id}")
+		result = cursor.fetchone()
+
+		return {
+			"name": result[0],
+			"manufacturer": result[1],
+			"price": result[2],
+			"image_name": result[3],
+			"image": base64.b64encode(result[4]).decode('utf-8')
+		}
+
+	@staticmethod
+	def create_user(username: str, email: str, password: str):
+		print("asd")

app/services/user_service.py

@@ -0,0 +1,129 @@
+import bcrypt
+import re
+import jwt
+import datetime
+from typing import Tuple, Union
+
+from app.extensions import db_cursor, db_connection
+from mysql.connector import Error
+
+from flask_jwt_extended import create_access_token
+
+
+class UserService:
+
+	@staticmethod
+	def create_user(username: str, email: str, password: str) -> Tuple[Union[dict, str], int]:
+
+		if not UserService.__verify_username(username):
+				return {"Failed": "Failed to verify username. Try another username"}, 400
+
+		if not UserService.__verify_email(email):
+				return {"Failed": "Failed to verify email. Try another email"}, 400
+
+		if not UserService.__verify_password(password):
+				return {"Failed": "Failed to verify password. Try another (stronger) password"}, 400
+
+		# Role ID 1 => Normal user
+		# Role ID 2 => Seller
+		# Role ID 3 => Admin
+		
+		hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())
+
+		try:
+			db_cursor.execute("select max(user_id) from user")
+			last_id = db_cursor.fetchone()[0]
+
+			if last_id < 23000:
+				return {"Failed": "Error occured when fetching last user id"}
+
+			new_id = last_id + 1
+
+			db_cursor.execute("insert into user (username, email, password, user_id, role_id) values (%s, %s, %s, %s, 1)", (username, email, hashed_password, new_id))
+			db_connection.commit()
+		except Error as e:
+			print(f"Error: {e}")
+			return {"Failed": "Failed to insert into database. Username or email are likely in use already"}, 500
+
+		return {"Success": "User created successfully"}, 200
+
+	@staticmethod
+	def login(username: str, password: str) -> Tuple[Union[dict, str], int]:
+
+		db_cursor.execute("select user_id, password, last_change from user where username = %s", (username,))
+		result = db_cursor.fetchone()
+
+		user_id = result[0]
+		password_hash = result[1]
+		last_change = result[2]
+
+		if user_id is None:
+			return {"Failed": "Username not found"}, 400
+
+		if not bcrypt.checkpw(password.encode('utf-8'), password_hash.encode('utf-8')):
+			return {"Failed": "Incorrect password"}, 401
+
+		expire = datetime.timedelta(days=1)
+
+		token = create_access_token(identity=user_id, expires_delta=expire,additional_claims={"lm": last_change})
+
+		return {"token": token}, 200
+
+	@staticmethod
+	def update_email(user_id: str, new_email: str) -> Tuple[Union[dict, str], int]:
+		
+		if not UserService.__verify_email(new_email):
+				return {"Failed": "Failed to verify email. Try another email"}, 400
+
+		try:
+			db_cursor.execute("update user set email = %s where user_id = %s", (new_email, user_id))
+			db_connection.commit()
+		except Error as e:
+			return {"Failed": f"Failed to update email. Email is likely in use already. Error: {e}"}, 500
+
+		return {"Success": "Email successfully updated"}, 200
+
+	@staticmethod
+	def update_username(user_id: str, new_username: str) -> Tuple[Union[dict, str], int]:
+		
+		if not UserService.__verify_username(new_username):
+			return {"Failed": "Failed to verify username. Try another one"}, 400
+
+		try:
+			db_cursor.execute("update user set username = %s where user_id = %s", (new_username, user_id))
+			db_connection.commit()
+		except Error as e:
+			return {"Failed": f"Failed to update username. Username is likely in use already. Error: {e}"}, 500
+
+		return {"Success": "Username successfully updated"}, 200
+
+	@staticmethod
+	def update_password(user_id: str, new_password: str) -> Tuple[Union[dict, str], int]:
+
+		if not UserService.__verify_password(new_password):
+			return {"Failed": "Failed to verify password. Try another (stronger) one"}, 400
+
+		hashed_password = bcrypt.hashpw(new_password.encode('utf-8'), bcrypt.gensalt())
+
+		try:
+			db_cursor.execute("update user set password = %s where user_id = %s", (new_username, user_id))
+			db_connection.commit()
+		except Error as e:
+			return {"Failed": f"Failed to update password. Error: {e}"}, 500
+
+		return {"Success": "Password successfully updated"}, 200
+
+	@staticmethod
+	def __verify_email(email: str) -> bool:
+		email_regex = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
+		return re.match(email_regex ,email) and len(email) <= 64
+
+	@staticmethod
+	def __verify_username(username: str) -> bool:
+		username_regex = r"^[a-zA-Z.-_]{1,64}$"
+		return re.match(username_regex, username)
+
+	@staticmethod
+	def __verify_password(password: str) -> bool:
+		password_regex = r"^(?=.*?[A-Z])(?=.*?[a-z])(?=.*?[0-9])(?=.*?[#?!@$ %^&*-]).{8,64}$"
+		return re.match(password_regex, password)

main.py

@@ -0,0 +1,13 @@
+from dotenv import load_dotenv
+from flask_jwt_extended import JWTManager
+import os
+
+from app import create_app
+
+load_dotenv()
+
+app = create_app()
+
+if __name__ == "__main__":
+	print("Hello, Flask")
+	app.run()

requirements.txt

@@ -0,0 +1,6 @@
+Flask==2.3.3
+gunicorn==20.1.0
+mysql-connector-python==8.3.0
+python-dotenv==1.0.1
+Flask-JWT-Extended==4.5.3
+PyJWT==2.8.0