mercredi 24 mai 2023

Code reusability issue when creating API with Flask

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 test

Aucun commentaire:

Enregistrer un commentaire