Mirrored_Repos/crafty-4
1
0
Fork 0
mirror of https://gitlab.com/crafty-controller/crafty-4.git synced 2024-08-30 18:23:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

cleaned up DB commands a bit to resolve the rest of CRAFTY-CONTROLLER-T-81

Browse Source
This commit is contained in:
computergeek125 2021-04-03 13:18:43 -05:00
parent 54b0578e79
commit f3b1d1e87d
6 changed files with 48 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
app/classes/shared/controller.py
View File

@ -345,7 +345,8 @@ class Controller:
self.stop_server(server_id) self.stop_server(server_id)
# remove the server from the DB # remove the server from the DB
#User_Servers.delete().where(User_Servers.server_id == server_id).execute() with database.atomic():
Role_Servers.delete().where(Role_Servers.server_id == server_id).execute()
Servers.delete().where(Servers.server_id == server_id).execute() Servers.delete().where(Servers.server_id == server_id).execute()
# remove the server from servers list # remove the server from servers list

66
app/classes/shared/models.py
View File

@ -327,31 +327,31 @@ class db_shortcuts:
return db_helper.return_rows(query) return db_helper.return_rows(query)
@staticmethod @staticmethod
def get_authorized_servers(userId): def get_authorized_servers(user_id):
userServers = User_Servers.select().where(User_Servers.user_id == userId) user_servers = User_Servers.select().where(User_Servers.user_id == user_id)
server_data = [] server_data = []
for u in userServers: for u in user_servers:
server_data.append(db_helper.get_server_data_by_id(u.server_id)) server_data.append(db_helper.get_server_data_by_id(u.server_id))
return server_data return server_data
@staticmethod @staticmethod
def get_authorized_servers_from_roles(userId): def get_authorized_servers_from_roles(user_id):
userRoles = User_Roles.select().where(User_Roles.user_id == userId) user_roles = User_Roles.select().where(User_Roles.user_id == user_id)
roles_list = [] roles_list = []
roleServer = [] role_server = []
server_data = [] server_data = []
for u in userRoles: for u in user_roles:
roles_list.append(db_helper.get_role(u.role_id)) roles_list.append(db_helper.get_role(u.role_id))
for r in roles_list: for r in roles_list:
role_test = Role_Servers.select().where(Role_Servers.role_id == r.get('role_id')) role_test = Role_Servers.select().where(Role_Servers.role_id == r.get('role_id'))
for t in role_test: for t in role_test:
roleServer.append(t) role_server.append(t)
for s in roleServer: for s in role_server:
server_data.append(db_helper.get_server_data_by_id(s.server_id)) server_data.append(db_helper.get_server_data_by_id(s.server_id))
return server_data return server_data
@ -367,40 +367,40 @@ class db_shortcuts:
return server_data return server_data
@staticmethod @staticmethod
def get_authorized_servers_stats(userId): def get_authorized_servers_stats(user_id):
userServers = User_Servers.select().where(User_Servers.user_id == userId) user_servers = User_Servers.select().where(User_Servers.user_id == user_id)
authorizedServers = [] authorized_servers = []
server_data = [] server_data = []
for u in userServers: for u in user_servers:
authorizedServers.append(db_helper.get_server_data_by_id(u.server_id)) authorized_servers.append(db_helper.get_server_data_by_id(u.server_id))
for s in authorizedServers: for s in authorized_servers:
latest = Server_Stats.select().where(Server_Stats.server_id == s.get('server_id')).order_by(Server_Stats.created.desc()).limit(1) latest = Server_Stats.select().where(Server_Stats.server_id == s.get('server_id')).order_by(Server_Stats.created.desc()).limit(1)
server_data.append({'server_data': s, "stats": db_helper.return_rows(latest)}) server_data.append({'server_data': s, "stats": db_helper.return_rows(latest)})
return server_data return server_data
@staticmethod @staticmethod
def get_authorized_servers_stats_from_roles(userId): def get_authorized_servers_stats_from_roles(user_id):
userRoles = User_Roles.select().where(User_Roles.user_id == userId) user_roles = User_Roles.select().where(User_Roles.user_id == user_id)
roles_list = [] roles_list = []
roleServer = [] role_server = []
authorizedServers = [] authorized_servers = []
server_data = [] server_data = []
for u in userRoles: for u in user_roles:
roles_list.append(db_helper.get_role(u.role_id)) roles_list.append(db_helper.get_role(u.role_id))
for r in roles_list: for r in roles_list:
role_test = Role_Servers.select().where(Role_Servers.role_id == r.get('role_id')) role_test = Role_Servers.select().where(Role_Servers.role_id == r.get('role_id'))
for t in role_test: for t in role_test:
roleServer.append(t) role_server.append(t)
for s in roleServer: for s in role_server:
authorizedServers.append(db_helper.get_server_data_by_id(s.server_id)) authorized_servers.append(db_helper.get_server_data_by_id(s.server_id))
for s in authorizedServers: for s in authorized_servers:
latest = Server_Stats.select().where(Server_Stats.server_id == s.get('server_id')).order_by(Server_Stats.created.desc()).limit(1) latest = Server_Stats.select().where(Server_Stats.server_id == s.get('server_id')).order_by(Server_Stats.created.desc()).limit(1)
server_data.append({'server_data': s, "stats": db_helper.return_rows(latest)[0]}) server_data.append({'server_data': s, "stats": db_helper.return_rows(latest)[0]})
return server_data return server_data
@ -417,9 +417,9 @@ class db_shortcuts:
return True return True
@staticmethod @staticmethod
def server_id_authorized(serverId, userId): def server_id_authorized(serverId, user_id):
userServer = User_Servers.select().where(User_Servers.server_id == serverId) userServer = User_Servers.select().where(User_Servers.server_id == serverId)
authorized = userServer.select().where(User_Servers.user_id == userId) authorized = userServer.select().where(User_Servers.user_id == user_id)
#authorized = db_helper.return_rows(authorized) #authorized = db_helper.return_rows(authorized)
if authorized.count() == 0: if authorized.count() == 0:
@ -427,14 +427,14 @@ class db_shortcuts:
return True return True
@staticmethod @staticmethod
def server_id_authorized_from_roles(serverId, userId): def server_id_authorized_from_roles(serverId, user_id):
cpt_authorized = 0 cpt_authorized = 0
roles_list = [] roles_list = []
roleServer = [] role_server = []
authorized = [] authorized = []
userRoles = User_Roles.select().where(User_Roles.user_id == userId) user_roles = User_Roles.select().where(User_Roles.user_id == user_id)
for u in userRoles: for u in user_roles:
roles_list.append(db_helper.get_role(u.role_id)) roles_list.append(db_helper.get_role(u.role_id))
for r in roles_list: for r in roles_list:
@ -472,7 +472,7 @@ class db_shortcuts:
return query return query
@staticmethod @staticmethod
def get_userid_by_name(username): def get_user_id_by_name(username):
if username == "SYSTEM": if username == "SYSTEM":
return 0 return 0
try: try:
@ -582,7 +582,7 @@ class db_shortcuts:
@staticmethod @staticmethod
def remove_user(user_id): def remove_user(user_id):
with database.atomic(): with database.atomic():
User_Roles.delete().where(User_Servers.user_id == user_id).execute() User_Roles.delete().where(User_Roles.user_id == user_id).execute()
user = Users.get(Users.user_id == user_id) user = Users.get(Users.user_id == user_id)
return user.delete_instance() return user.delete_instance()
@ -650,7 +650,9 @@ class db_shortcuts:
@staticmethod @staticmethod
def remove_role(role_id): def remove_role(role_id):
with database.atomic():
Role_Servers.delete().where(Role_Servers.role_id == role_id).execute() Role_Servers.delete().where(Role_Servers.role_id == role_id).execute()
User_Roles.delete().where(User_Roles.role_id == role_id).execute()
role = Roles.get(Roles.role_id == role_id) role = Roles.get(Roles.role_id == role_id)
return role.delete_instance() return role.delete_instance()

6
app/classes/web/base_handler.py
View File

@ -12,6 +12,8 @@ logger = logging.getLogger(__name__)
class BaseHandler(tornado.web.RequestHandler): class BaseHandler(tornado.web.RequestHandler):
nobleach = {bool, type(None)}
def initialize(self, controller=None, tasks_manager=None, translator=None): def initialize(self, controller=None, tasks_manager=None, translator=None):
self.controller = controller self.controller = controller
self.tasks_manager = tasks_manager self.tasks_manager = tasks_manager
@ -27,10 +29,10 @@ class BaseHandler(tornado.web.RequestHandler):
return self.get_secure_cookie("user", max_age_days=1) return self.get_secure_cookie("user", max_age_days=1)
def autobleach(self, text): def autobleach(self, text):
if type(text) is bool: if type(text) in self.nobleach:
return text return text
else: else:
return text return bleach.clean(text)
def get_argument( def get_argument(
self, self,

2
app/classes/web/default_handler.py
View File

@ -18,6 +18,6 @@ class DefaultHandler(BaseHandler):
else: else:
self.redirect( self.redirect(
"/public/login", "/public/login",
translate=self.translator.translate, #translate=self.translator.translate,
) )

4
app/classes/web/panel_handler.py
View File

@ -373,7 +373,7 @@ class PanelHandler(BaseHandler):
return return
else: else:
# does this user id exist? # does this user id exist?
target_role = db_helper.get_user(role_id) target_role = db_helper.get_role(role_id)
if not target_role: if not target_role:
self.redirect("/panel/error?error=Invalid Role ID") self.redirect("/panel/error?error=Invalid Role ID")
return return
@ -582,7 +582,7 @@ class PanelHandler(BaseHandler):
) )
)) ))
if argument: if argument:
roles.add(role['role_id']) roles.add(role.role_id)
servers = set() servers = set()
for server in self.controller.list_defined_servers(): for server in self.controller.list_defined_servers():

2
app/classes/web/server_handler.py
View File

@ -50,7 +50,7 @@ class ServerHandler(BaseHandler):
page_data = { page_data = {
'version_data': helper.get_version_string(), 'version_data': helper.get_version_string(),
'user_data': exec_user_data, 'user_data': exec_user_data,
'user_role' : user_role, 'user_role' : exec_user_role,
'server_stats': { 'server_stats': {
'total': len(self.controller.list_defined_servers()), 'total': len(self.controller.list_defined_servers()),
'running': len(self.controller.list_running_servers()), 'running': len(self.controller.list_running_servers()),