I have code reusability issue in my code, this is one of example of routes folder file, it's users.py:
from models import User, ApiResponse
from mongoengine import ValidationError
from flask import request
from utils import get_timestamp
from copy import deepcopy
from authentication import check_password_hash, generate_password_hash, tokenizer, detokenizer, get_data_by_token
from validation import is_identity_valid, is_role_valid
from env import ENV
def create_user():
try:
# Authentication check
auth = request.cookies.get('auth') # type: ignore
if auth is None:
return ApiResponse(401, 'fail', 'Unauthorized, auth is none.', None)
data = get_data_by_token(str(auth))
if data is None:
return ApiResponse(401, 'fail', 'Unauthorized, token is none.', None)
user, newToken = data
if user is None:
return ApiResponse(401, 'fail', 'Unauthorized, user is none.', None)
if not str(user.role) in ['admin']:
return ApiResponse(403, 'fail', 'Previlege is forbidden.', None)
# Get timestamp
# 1683478800000 adalah timestamp proyek ini dimulai
timestamp = get_timestamp()
body = request.json
if body:
userReq = User(**body['data']) # JSON deserialize
if not is_identity_valid(str(userReq.identity)):
return ApiResponse(400, 'fail', 'Invalid identity.', None)
if not is_role_valid(str(userReq.role)):
print(userReq.role)
return ApiResponse(400, 'fail', 'Invalid role.', None)
# Mencari user yang ada di database
userDbList = User.objects(identity=userReq.identity) # type: ignore
if len(userDbList) != 0:
return ApiResponse(409, 'fail', 'User already exists.', None)
userReq._id = timestamp
userReq.password = generate_password_hash(str(userReq.password))
userReq.save()
resp = ApiResponse(
201, 'success', 'User created successfully.', userReq.to_mongo())
if newToken:
resp.set_cookie('auth', newToken, path=ENV.API_ENDPOINT)
return resp
else:
return ApiResponse(400, 'fail', 'Bad request.', None)
except ValidationError:
return ApiResponse(422, 'fail', 'Unprocessable entity.', None)
except Exception as e:
print(e)
return ApiResponse(500, 'fail', 'Internal server error.', None)
def get_users():
try:
# Authentication check
auth = request.cookies.get('auth') # type: ignore
if auth is None:
return ApiResponse(401, 'fail', 'Unauthorized, auth is none.', None)
data = get_data_by_token(str(auth))
if data is None:
return ApiResponse(401, 'fail', 'Unauthorized, token is none.', None)
user, newToken = data
if user is None:
return ApiResponse(401, 'fail', 'Unauthorized, user is none.', None)
if not str(user.role) in ['admin']:
return ApiResponse(403, 'fail', 'Previlege is forbidden.', None)
users = User.objects().all() # type: ignore
userList = [user.to_mongo() for user in users]
resp = ApiResponse(
200, 'success', 'Users retrieved successfully.', userList)
resp.set_cookie('auth', newToken, path=ENV.API_ENDPOINT)
return resp
except Exception as e:
print(e)
return ApiResponse(500, 'fail', 'Internal server error.', None)
def get_user(user_id):
try:
# Authentication check
auth = request.cookies.get('auth') # type: ignore
if auth is None:
return ApiResponse(401, 'fail', 'Unauthorized, auth is none.', None)
data = get_data_by_token(str(auth))
if data is None:
return ApiResponse(401, 'fail', 'Unauthorized, token is none.', None)
user, newToken = data
if user is None:
return ApiResponse(401, 'fail', 'Unauthorized, user is none.', None)
# Admin check
if not str(user.role) in ['admin']:
if int(user._id) != int(user_id): # type: ignore
print(int(user._id), int(user_id)) # type: ignore
return ApiResponse(403, 'fail', 'Forbidden.', None)
userList = User.objects(_id=int(user_id)) # type: ignore
if len(userList) == 0:
return ApiResponse(404, 'fail', 'User not found.', None)
userReq = userList[0]
resp = ApiResponse(
200, 'success', 'User retrieved successfully.', userReq.to_mongo())
resp.set_cookie('auth', newToken, path=ENV.API_ENDPOINT)
return resp
except Exception as e:
print(e)
return ApiResponse(500, 'fail', 'Internal server error.', None)
def put_user(user_id):
try:
# Authentication check
auth = request.cookies.get('auth') # type: ignore
if auth is None:
return ApiResponse(401, 'fail', 'Unauthorized, auth is none.', None)
data = get_data_by_token(str(auth))
if data is None:
return ApiResponse(401, 'fail', 'Unauthorized, token is none.', None)
user, newToken = data
if user is None:
return ApiResponse(401, 'fail', 'Unauthorized, user is none.', None)
# Admin check
isAdmin = True
if not str(user.role) in ['admin']:
if int(user._id) != int(user_id): # type: ignore
return ApiResponse(403, 'fail', 'Forbidden.', None)
isAdmin = False
userList = User.objects(_id=int(user_id)) # type: ignore
if len(userList) == 0:
return ApiResponse(404, 'fail', 'User not found.', None)
userOld = userList[0]
body = request.json
if body:
oldUser = deepcopy(userOld)
newUser = User(**body['data'])
# Field update
if newUser.password is not None:
userOld.password = generate_password_hash(str(newUser.password))
if is_role_valid(str(newUser.role)) and isAdmin:
userOld.role = str(newUser.role)
userOld.save()
response = {
'old_data': oldUser.to_mongo(),
'new_data': userOld.to_mongo()
}
resp = ApiResponse(200, 'success', 'User updated succesfully.', response)
resp.set_cookie('auth', newToken, path=ENV.API_ENDPOINT)
return resp
else:
return ApiResponse(400, 'fail', 'Bad request.', None)
except Exception as e:
print(e)
return ApiResponse(500, 'fail', 'Internal server error.', None)
def delete_user(user_id):
try:
# Authentication check
auth = request.cookies.get('auth') # type: ignore
if auth is None:
return ApiResponse(401, 'fail', 'Unauthorized, auth is none.', None)
data = get_data_by_token(str(auth))
if data is None:
return ApiResponse(401, 'fail', 'Unauthorized, token is none.', None)
user, newToken = data
if user is None:
return ApiResponse(401, 'fail', 'Unauthorized, user is none.', None)
# Admin check
if not str(user.role) in ['admin']:
return ApiResponse(403, 'fail', 'Forbidden.', None)
# Fetch the user from the collection based on the provided ID
userList = User.objects(_id=int(user_id)) # type: ignore
# Check if user found
if len(userList) == 0:
return ApiResponse(404, 'fail', 'User not found.', None)
userReq = userList[0]
# Remove from database
userReq.delete()
resp = ApiResponse(200, 'success', 'User deleted successfully.', None)
resp.set_cookie('auth', newToken, path=ENV.API_ENDPOINT)
return resp
except Exception as e:
print(e)
return ApiResponse(500, 'fail', 'Internal server error.', None)
def login_user():
try:
body = request.json
if body:
userReq = User(**body['data'])
userList = User.objects(identity=str(userReq.identity)) # type: ignore
if len(userList) == 0:
return ApiResponse(401, 'fail', 'Wrong authentication.', None)
user = userList[0]
if not check_password_hash(str(user.password), str(userReq.password)):
return ApiResponse(401, 'fail', 'Wrong authentication.', None)
# Authorization return token
dataForToken = {
'identity': user.identity
}
token = tokenizer(dataForToken)
if token is None:
raise Exception('Token generating error.')
user.save()
resp = ApiResponse(
200, 'success', f'User {user.identity} logged in successfully.', token)
resp.set_cookie('auth', token, path=ENV.API_ENDPOINT)
return resp
else:
return ApiResponse(400, 'fail', 'Bad request.', None)
except Exception as e:
print(e)
return ApiResponse(500, 'fail', 'Internal server error.', None)
def logout_user():
try:
# Authentication check
auth = request.cookies.get('auth') # type: ignore
if auth is None:
return ApiResponse(401, 'fail', 'Unauthorized, auth is none.', None)
data = get_data_by_token(str(auth))
if data is None:
return ApiResponse(401, 'fail', 'Unauthorized, token is none.', None)
user, newToken = data
if user is None:
return ApiResponse(401, 'fail', 'Unauthorized, user is none.', None)
if not str(user.role) in ['admin']:
return ApiResponse(403, 'fail', 'Previlege is forbidden.', None)
# Deauthentication
resp = ApiResponse(
200, 'success', f'User {user.identity} logged out successfully.', None)
resp.set_cookie('auth', '', path=ENV.API_ENDPOINT)
return resp
except Exception as e:
print(e)
return ApiResponse(500, 'fail', 'Internal server error.', None)
I'm struggle finding way how to keep it simple of every route. Mainly cookie store.
Let me know if you want to know objects or class from my code example.
I think I have really bad design pattern. Here is my project structure