41 lines
1.1 KiB
Plaintext
41 lines
1.1 KiB
Plaintext
async def checkUserPermission(user, path):
|
|
if user is None:
|
|
user = 'anonymous_user'
|
|
# print(f'kboss check permission {user} for {path}')
|
|
|
|
sql = """select distinct a.path, d.id
|
|
from permission a
|
|
left join rolepermission b on a.id = b.permid
|
|
left join userrole c on c.roleid = b.roleid
|
|
left join users d on d.id = c.userid
|
|
where a.path = ${path}$
|
|
and a.del_flg = '0'
|
|
and (b.del_flg = '0' or b.del_flg is NULL)
|
|
and (c.del_flg = '0' or c.del_flg is NULL)
|
|
and (d.del_flg = '0' or d.del_flg is NULL)
|
|
and (d.user_status = '0' or d.user_status is NULL)
|
|
-- and (d.id = ${user}$ or d.username is Null)
|
|
"""
|
|
|
|
db = DBPools()
|
|
dbname = 'kboss'
|
|
# print(f'database name is {dbname}')
|
|
async with db.sqlorContext(dbname) as sor:
|
|
recs = await sor.sqlExe(sql, {'path':path, 'user':user})
|
|
if len(recs) == 0:
|
|
return f'{user=}, {path=} not found in permission'
|
|
for r in recs:
|
|
id = r['id']
|
|
if id == user:
|
|
return f'{user=} has {path=} permission, {id=}'
|
|
return {
|
|
'user':user,
|
|
'recs':recs
|
|
}
|
|
return f'{user} has not permission to call {path}'
|
|
|
|
user = params_kw.get('userid')
|
|
path = params_kw.get('path')
|
|
return await checkUserPermission(user, path)
|
|
|