open-RESTful-API/BackendApi/SharixQuery.py

from FlaskApp.BackendApi.Models import RoleModel, UserModel
from FlaskApp.BackendApi.connect import connect
from sqlalchemy import create_engine,  or_
from sqlalchemy.orm import sessionmaker, scoped_session


engine = create_engine(connect())

session = scoped_session(sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
))

print(str(session.is_active))

class ReturnData:
    def __init__(self, data, code):
        self.data = data
        self.code = code
class SharixQuery:
    
    def auth_query(self, request_data) -> ReturnData:
        try:
            result = session.query(UserModel).\
                filter(or_(UserModel.email == request_data['login_email'], UserModel.name == request_data['login_email'])).\
                filter(UserModel.password == request_data['password']).first()
            return ReturnData(result, 200)
        except Exception as ex:
            return ReturnData({"msg":str(ex)}, 500)
    def add_user(self, request_data) -> ReturnData:
        try:
            user = UserModel(**request_data)
            chek = session.query(UserModel).filter(or_(UserModel.email==user.email, UserModel.name==user.name)).first()
            if chek is None:
                session.add(user)
                session.flush()
                session.commit()
                return ReturnData({"id":user.id}, 200)
            else:
                return ReturnData({"msg":"User exist!"}, 400)
        except Exception as ex:
            return ReturnData({"msg":str(ex)}, 500)
    
    def get_user(self, id) -> UserModel:
        try:
            user = session.query(UserModel).filter(UserModel.id == id).first()
            return user
        except Exception as ex:
            return ReturnData({"msg":str(ex)}, 500)    
    
    def get_all_user(self) -> list[UserModel]:
        try:
            alluser = session.query(UserModel).order_by(UserModel.email).all()
            return alluser
        except Exception as ex:
            return ReturnData({"msg":str(ex)}, 500)
    
    def delete_user_query(self, filed, filed_value) -> ReturnData:
        try:
            session.query(UserModel).filter(filed == filed_value).delete()
            session.commit()
            return ReturnData({"msg":"User is deleted!"}, 200)
        except Exception as ex:
            return ReturnData({"msg":str(ex)}, 500)
    
    def update_user(self, request_data) -> ReturnData:
        try:
            user = UserModel(**request_data)
            session.query(UserModel).filter(UserModel.id == user.id).update({
                UserModel.name:UserModel.name if user.name is None else user.name,
                UserModel.password:UserModel.password if user.password is None else user.password,
                })
            session.flush()
            session.commit()
            return ReturnData(user.as_dict(), 200)
        except Exception as ex:
            return ReturnData({"msg":str(ex)}, 500)
    
    def get_roles(self) -> list[RoleModel]:
        try:
            all_role = session.query(RoleModel).all()
            return all_role
        except Exception as ex:
            return ReturnData({"msg":str(ex)}, 500)
    
    def delete_role(self, filed, filed_value) -> ReturnData:
        try:
            self.delete_user_query(UserModel.idrole, filed_value)
            session.query(RoleModel).filter(filed == filed_value).delete()
            session.commit()
            return ReturnData({"msg":"Role is deleted!"}, 200)
        except Exception as ex:
            return ReturnData({"msg":str(ex)}, 500)
    
    def add_role(self, request_data) -> ReturnData:
        try:
            role = RoleModel(**request_data)
            session.add(role)
            session.flush()
            session.commit()
            return ReturnData({"id":str(role.id)}, 500)
        except Exception as ex:
            return ReturnData({"msg":str(ex)}, 500)