aji/costapy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add roles modules and add it on handler

Browse Source
This commit is contained in:
Dita Aji Pratama 2025-04-25 13:55:59 +07:00
parent 345ad468a3
commit a4887a2286
2 changed files with 165 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
handler.py
View File

@ -22,6 +22,7 @@ import modules.public.forgot as public_forgot
import modules.public.reset as public_reset import modules.public.reset as public_reset
import modules.api.auth as api_auth import modules.api.auth as api_auth
import modules.api.roles as api_auth_roles
app = Bottle() app = Bottle()
@ -247,3 +248,24 @@ def index():
print(str(e),flush=True) print(str(e),flush=True)
return json.dumps({}, indent = 2).encode() return json.dumps({}, indent = 2).encode()
@app.route('/api/auth/roles/<alder>', method=['OPTIONS', 'POST'])
def index(alder):
try:
if request.method == 'OPTIONS':
return None
else:
response.content_type = 'application/json'
params = request.json
if alder == 'add':
return json.dumps(api_auth_roles.roles().add(params), indent = 2).encode()
elif alder == 'list':
return json.dumps(api_auth_roles.roles().list(params), indent = 2).encode()
elif alder == 'edit':
return json.dumps(api_auth_roles.roles().edit(params), indent = 2).encode()
elif alder == 'remove':
return json.dumps(api_auth_roles.roles().remove(params), indent = 2).encode()
else:
return json.dumps({}, indent = 2).encode()
except Exception as e:
print(str(e),flush=True)
return json.dumps({}, indent = 2).encode()

143
modules/api/roles.py Normal file
View File

@ -0,0 +1,143 @@
import mysql.connector as mariadb
from mako.template import Template
from bottle import request
from config import database, globalvar
from scripts import loggorilla, tokenguard
import procedure.validation as procedure_validation
class roles:
def __init__(self):
self.db_main = mariadb.connect(**database.db_main)
self.cursor = self.db_main.cursor(dictionary=True)
def add(self, params):
APIADDR = "/api/auth/roles/add"
response = {}
loggorilla.prcss(APIADDR, "Define parameters")
token = params["token" ]
id = params["id" ]
name = params["name" ]
allowed_roles = [1]
loggorilla.prcss(APIADDR, "Account validation")
user_validation = procedure_validation.validation().account(APIADDR, allowed_roles, token)
user = user_validation['data']
self.cursor.execute("BEGIN;")
try:
self.cursor.execute("INSERT INTO `auth_roles` VALUES (%s, %s) ;", (id, name) )
response["status" ] = "success"
response["desc" ] = "data added"
except Exception as e:
self.cursor.execute("ROLLBACK;")
loggorilla.error(APIADDR, str(e) )
response["status" ] = "failed"
response["desc" ] = "Internal Server Error. Please contact us if you still have an error."
finally:
self.cursor.execute("COMMIT;")
self.cursor.close()
self.db_main.close()
return response
def list(self, params):
APIADDR = "/api/auth/roles/list"
response = {}
loggorilla.prcss(APIADDR, "Define parameters")
token = params["token" ]
allowed_roles = [1,2] # Roles list is public or not?
loggorilla.prcss(APIADDR, "Account validation")
user_validation = procedure_validation.validation().account(APIADDR, allowed_roles, token)
user = user_validation['data']
self.cursor.execute("BEGIN;")
try:
self.cursor.execute("select auth_roles.id, auth_roles.name, (select count(*) from auth_profile_roles apr where apr.roles = auth_roles.id) AS `count` from auth_roles;")
r_roles = self.cursor.fetchall()
response["status" ] = "success"
response["desc" ] = "data collected"
response["data" ] = r_roles
except Exception as e:
self.cursor.execute("ROLLBACK;")
loggorilla.error(APIADDR, str(e) )
response["status" ] = "failed"
response["desc" ] = "Internal Server Error. Please contact us if you still have an error."
finally:
self.cursor.execute("COMMIT;")
self.cursor.close()
self.db_main.close()
return response
def edit(self, params):
APIADDR = "/api/auth/roles/edit"
response = {}
loggorilla.prcss(APIADDR, "Define parameters")
token = params["token" ]
key = params["key" ]
id = params["id" ]
name = params["name" ]
allowed_roles = [1]
loggorilla.prcss(APIADDR, "Account validation")
user_validation = procedure_validation.validation().account(APIADDR, allowed_roles, token)
user = user_validation['data']
self.cursor.execute("BEGIN;")
try:
if key == 1 or id == 1:
response["status" ] = "failed"
response["desc" ] = "Cannot change super user"
else:
self.cursor.execute("UPDATE `auth_roles` SET `id` = %s, `name` = %s WHERE `id` = %s ;", (id, name, key) )
response["status" ] = "success"
response["desc" ] = "data change"
except Exception as e:
self.cursor.execute("ROLLBACK;")
loggorilla.error(APIADDR, str(e) )
response["status" ] = "failed"
response["desc" ] = "Internal Server Error. Please contact us if you still have an error."
finally:
self.cursor.execute("COMMIT;")
self.cursor.close()
self.db_main.close()
return response
def remove(self, params):
APIADDR = "/api/auth/roles/remove"
response = {}
loggorilla.prcss(APIADDR, "Define parameters")
token = params["token" ]
key = params["key" ]
allowed_roles = [1]
loggorilla.prcss(APIADDR, "Account validation")
user_validation = procedure_validation.validation().account(APIADDR, allowed_roles, token)
user = user_validation['data']
self.cursor.execute("BEGIN;")
try:
if key == 1:
response["status" ] = "failed"
response["desc" ] = "Cannot change super user"
else:
self.cursor.execute("DELETE FROM `auth_roles` WHERE `id` = %s ;", (key,) )
response["status" ] = "success"
response["desc" ] = "data removed"
except Exception as e:
self.cursor.execute("ROLLBACK;")
loggorilla.error(APIADDR, str(e) )
response["status" ] = "failed"
response["desc" ] = "Internal Server Error. Please contact us if you still have an error."
finally:
self.cursor.execute("COMMIT;")
self.cursor.close()
self.db_main.close()
return response