add back

BackendApi/Models.py

@@ -0,0 +1,26 @@
+from sqlalchemy import Boolean, Column, Date, DateTime, Integer, String
+from sqlalchemy.orm import declarative_base
+
+Base = declarative_base()
+
+class UserModel(Base):
+    __tablename__ = "rest_user"
+    id = Column(Integer, primary_key=True, autoincrement=True, )
+    name = Column(String)
+    birthday = Column(Date)
+    insys = Column(Boolean)
+    idrole = Column(Integer)
+    lastlogintime = Column(DateTime)
+    email = Column(String)
+    password = Column(String)
+
+    def as_dict(self):
+        return {c.name: getattr(self, c.name) for c in self.__table__.columns}
+
+class RoleModel(Base):
+    __tablename__ = 'rest_role'
+    id = Column(Integer, primary_key=True, autoincrement=True, )
+    name = Column(String, unique=True, nullable=False)
+
+    def as_dict(self):
+        return {c.name: getattr(self, c.name) for c in self.__table__.columns}

BackendApi/SharixQuery.py

@@ -0,0 +1,105 @@
+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)
+ 

BackendApi/config.py

@@ -0,0 +1,4 @@
+host = "localhost"
+user = "postgres"
+password = ""
+db_name = "postgres"

BackendApi/connect.py

@@ -0,0 +1,11 @@
+import psycopg2 as pg
+from FlaskApp.BackendApi.config import host, user, db_name, password
+
+def connect():
+	try:
+		print('Connecting to the PostgreSQL Database...')
+		conn1 = f"postgresql://postgres@192.168.1.98/postgres"
+		print("Success database connect!")
+		return conn1
+	except:
+		print('Connection Has Failed...')