617 lines
25 KiB
Python
617 lines
25 KiB
Python
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from motor.motor_asyncio import AsyncIOMotorDatabase
|
|
|
|
import crud
|
|
import schemas
|
|
from api import deps
|
|
from db import get_database
|
|
from db.ckdb import CKDrive, get_ck_db
|
|
from db.redisdb import RedisDrive, get_redis_pool
|
|
from models.behavior_analysis import BehaviorAnalysis
|
|
from utils import casbin_enforcer
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/add_role_domain")
|
|
async def add_role_domain(
|
|
request: Request,
|
|
data_in: schemas.AddRoleForUsersInDomain,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)):
|
|
"""
|
|
在当前项目为角色添加相应权限
|
|
"""
|
|
|
|
# username role dom
|
|
# for item in data_in.data:
|
|
# is_exists_role = await crud.role.check(db, _id=item.role_id, game=item.game)
|
|
# if not is_exists_role:
|
|
# continue
|
|
# casbin_enforcer.add_role_for_user_in_domain(user=item.username,
|
|
# role=item.role_id,
|
|
# domain=item.game)
|
|
#
|
|
# return schemas.Msg(code=0, msg='添加成功', data=True)
|
|
res = await crud.url_list.get_all(db) # 获取所有级别权限的所有路由和路由状态
|
|
role_id = {}
|
|
for i in res:
|
|
role_id[i['auth_id']] = i['name']
|
|
for item in data_in.data:
|
|
now_quanxian = await crud.user_url.get_quanxian(db, schemas.Url_quanxian(user_id=item.role_id))
|
|
# 如果不存在该用户其他游戏的权限，则新增一个
|
|
if now_quanxian == {}:
|
|
await crud.user_url.insert_quanxian(db, schemas.Url_quanxian(game=[item.game], user=item.username,
|
|
user_id=item.role_id,
|
|
quanxian=[role_id[item.auth_id]],
|
|
quanxian_id=[item.auth_id]))
|
|
# 存在则在这个用户加上要添加的游戏项目权限
|
|
else:
|
|
game = now_quanxian['game']
|
|
game.append(item.game)
|
|
quanxian = now_quanxian['quanxian']
|
|
quanxian.append(role_id[item.auth_id])
|
|
quanxian_id = now_quanxian['quanxian_id']
|
|
quanxian_id.append('auth_id')
|
|
await crud.user_url.updata_quanxian(db, schemas.Url_quanxian(game=game, user=item.username,
|
|
user_id=item.role_id, quanxian=quanxian,
|
|
quanxian_id=quanxian_id))
|
|
return schemas.Msg(code=0, msg='添加成功', data=True)
|
|
|
|
# 疑似弃用
|
|
@router.post("/get_permissions_for_user_in_domain")
|
|
async def get_permissions_for_user_in_domain(
|
|
request: Request,
|
|
data_in: schemas.GetPermissionsForUserInDomain,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)):
|
|
"""
|
|
获取域内用户或角色的权限
|
|
"""
|
|
#data为列表
|
|
data = casbin_enforcer.get_permissions_for_user_in_domain(data_in.role_id, data_in.game)
|
|
paths = {i[2] for i in data}
|
|
#列表形式的coll_name
|
|
all_api = await crud.api_list.all_api(db)
|
|
for item in all_api:
|
|
if item['path'] in paths:
|
|
item['is_authz'] = True
|
|
else:
|
|
item['is_authz'] = False
|
|
|
|
return schemas.Msg(code=0, msg='ok', data=all_api)
|
|
|
|
|
|
@router.post("/del_role_user_domain")
|
|
async def del_role_domain(
|
|
request: Request,
|
|
data_in: schemas.DeleteRolesForUserInDomain,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)):
|
|
"""
|
|
删除用户在当前项目中的权限
|
|
"""
|
|
|
|
# username role dom
|
|
# res = casbin_enforcer.delete_roles_for_user_in_domain(user=data_in.username,
|
|
# role=data_in.role_id,
|
|
# domain=data_in.game)
|
|
#
|
|
# #await crud.role.delete_id(db, data_in.role_id)
|
|
# return schemas.Msg(code=0, msg='ok', data=res)
|
|
res = await crud.user_url.get_all(db)
|
|
for i in res:
|
|
if i['user'] == data_in.username:
|
|
for nu in range(len(i['game'])):
|
|
if i['game'][nu] == data_in.game:
|
|
i['game'].remove(data_in.game)
|
|
i['quanxian_id'].remove(i['quanxian_id'][nu])
|
|
i['quanxian'].remove(data_in.role_id)
|
|
await crud.user_url.updata_quanxian(db, schemas.Url_quanxian(game=i['game'], user=data_in.username,
|
|
user_id=i['user_id'],
|
|
quanxian_id=i['quanxian_id'],
|
|
quanxian=i['quanxian']))
|
|
return schemas.Msg(code=0, msg='删除成功', data='')
|
|
|
|
|
|
@router.post("/del_role_user")
|
|
async def del_role_domain(
|
|
request: Request,
|
|
data_in: schemas.DeleteRolesForUserInDomain,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)):
|
|
"""
|
|
删除角色管理板块中的角色
|
|
"""
|
|
await crud.url_list.delete_name(db, data_in)
|
|
return schemas.Msg(code=0, msg="ok", data='')
|
|
|
|
|
|
@router.post("/add_policy")
|
|
async def add_policy(
|
|
request: Request,
|
|
data_in: schemas.Datalist,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)):
|
|
"""
|
|
向当前权限添加新路由
|
|
"""
|
|
# res = 0
|
|
# for path in data_id.path_list:
|
|
# res = casbin_enforcer.add_policy(data_id.role_id, data_id.game, path, data_id.act)
|
|
# return schemas.Msg(code=0, msg='ok', data=res)
|
|
res = await crud.url_list.find_one_url(db, data_in)
|
|
for i in range(len(res['api_list'])):
|
|
if res['api_list'][i] == data_in.path:
|
|
res['state'][i] = True
|
|
await crud.url_list.update_url_url(db, res)
|
|
return schemas.Msg(code=0, msg='修改成功', data='')
|
|
|
|
@router.post("/del_policy")
|
|
async def remove_policy(
|
|
request: Request,
|
|
data_in: schemas.Del_role,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)):
|
|
"""
|
|
修改角色api权限
|
|
"""
|
|
# res = casbin_enforcer.remove_policy(data_id.role_id, data_id.game, data_id.path, data_id.act)
|
|
# return schemas.Msg(code=0, msg='ok', data=res)
|
|
res = await crud.url_list.find_one_url(db, data_in)
|
|
for i in range(len(res['api_list'])):
|
|
if res['api_list'][i] == data_in.path:
|
|
res['state'][i] = False
|
|
await crud.url_list.update_url_url(db, res)
|
|
return schemas.Msg(code=0, msg='修改成功', data='')
|
|
|
|
|
|
@router.post("/del_api_module")
|
|
async def add_policy(
|
|
request: Request,
|
|
data_in: schemas.Add_module,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)):
|
|
res = await crud.api_module.get_one_module(db, data_in)
|
|
for i in range(len(res['state'])):
|
|
if data_in.url == res['api_list'][i]:
|
|
res['state'][i] = False
|
|
await crud.api_module.update_one_module(db, res)
|
|
return schemas.Msg(code=0, msg='修改成功', data='')
|
|
|
|
|
|
@router.post("/add_api_module")
|
|
async def add_policy(
|
|
request: Request,
|
|
data_in: schemas.Add_module,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)):
|
|
res = await crud.api_module.get_one_module(db, data_in)
|
|
for i in range(len(res['state'])):
|
|
if data_in.url == res['api_list'][i]:
|
|
res['state'][i] = True
|
|
await crud.api_module.update_one_module(db, res)
|
|
return schemas.Msg(code=0, msg='修改成功', data='')
|
|
|
|
|
|
@router.get("/api_list")
|
|
async def api_list(
|
|
request: Request,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)):
|
|
"""
|
|
|
|
GetPermissionsForUserInDomain
|
|
所有的api
|
|
"""
|
|
# res = await crud.api_list.all_api(db)
|
|
# return schemas.Msg(code=0, msg='ok', data=res)
|
|
re = await crud.api_module.get_api_module(db)
|
|
res = []
|
|
for i in re:
|
|
if i['path_name'] != 'root':
|
|
i['_id'] = str(i['_id'])
|
|
res.append(i)
|
|
return schemas.Msg(code=0, msg='ok', data=res)
|
|
|
|
@router.post("/add_api")
|
|
async def add_api(
|
|
request: Request,
|
|
data_in: schemas.AddApi,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)
|
|
) -> schemas.Msg:
|
|
"""
|
|
添加api
|
|
"""
|
|
# try:
|
|
# res = await crud.api_list.add_api(db, data_in)
|
|
# except Exception as e:
|
|
# return schemas.Msg(code=-1, msg='已经存在')
|
|
# return schemas.Msg(code=0, msg='ok', data=res.matched_count)
|
|
res = await crud.api_module.get_api_module(db)
|
|
for i in res:
|
|
if data_in.path in i['api_list']:
|
|
return schemas.Msg(code=0, msg='该路由已存在', data='')
|
|
path_list = []
|
|
for i in res:
|
|
path_list.append(i['path_name'])
|
|
if data_in.name in path_list:
|
|
for i in res:
|
|
if data_in.name == i['path_name']:
|
|
i['api_list'].append(data_in.path)
|
|
i['api_name'].append(data_in.desc)
|
|
i['state'].append(True)
|
|
await crud.api_module.updata_quanxian_module(db, schemas.Url_module(auth_id=i['auth_id'],
|
|
path_name=data_in.name,
|
|
api_list=i['api_list'],
|
|
api_name=i['api_name'],
|
|
state=i['state']))
|
|
return schemas.Msg(code=0, msg='ok', data='路由添加成功！')
|
|
else:
|
|
auth_list = []
|
|
for i in res:
|
|
auth_list.append(i['auth_id'])
|
|
auth_id = max(auth_list)
|
|
# api_data={}
|
|
# api_data['auth_id']='abc'+str(int(auth_id.split('c')[-1])+1)
|
|
# api_data['path_name']=data_in.name
|
|
# api_data['api_list']=[data_in.path]
|
|
# api_data['api_name']=[data_in.desc]
|
|
# api_data['state']=[True]
|
|
auth_id = 'abc' + str(int(auth_id.split('c')[-1]) + 1)
|
|
await crud.api_module.insert_quanxian(db, schemas.Url_module(auth_id=auth_id, path_name=data_in.name,
|
|
api_list=[data_in.path],
|
|
api_name=[data_in.desc], state=[True]))
|
|
return schemas.Msg(code=0, msg='ok', data='路由添加成功！')
|
|
|
|
|
|
@router.post("/del_api")
|
|
async def del_api(
|
|
request: Request,
|
|
data_in: schemas.DelApi,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)) -> schemas.Msg:
|
|
"""
|
|
删除api
|
|
"""
|
|
# 删除规则
|
|
paths = await crud.api_list.find_ids(db, data_in.ids, {'path': 1})
|
|
for item in paths:
|
|
casbin_enforcer.remove_filtered_policy(2, item['path'])
|
|
|
|
# 删除保存的记录
|
|
res = await crud.api_list.del_api(db, data_in)
|
|
|
|
return schemas.Msg(code=0, msg='ok', data=res.deleted_count)
|
|
|
|
|
|
@router.post("/edit_api")
|
|
async def edit_api(
|
|
request: Request,
|
|
data_in: schemas.EditApi,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)) -> schemas.Msg:
|
|
"""
|
|
编辑api
|
|
"""
|
|
res = await crud.api_list.edit_api(db, data_in)
|
|
return schemas.Msg(code=0, msg='ok', data=res.matched_count)
|
|
|
|
|
|
@router.get("/domain")
|
|
async def domain_list(
|
|
request: Request,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)
|
|
) -> schemas.Msg:
|
|
"""
|
|
获取所有项目
|
|
"""
|
|
# roel dom path *
|
|
res = await crud.project.all_game(db)
|
|
return schemas.Msg(code=0, msg='ok', data=res)
|
|
@router.get("/api_module")
|
|
async def domain_list(
|
|
request: Request,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)
|
|
) -> schemas.Msg:
|
|
"""
|
|
角色管理创建角色时显示的各个模块
|
|
"""
|
|
res = await crud.api_module.get_api_module(db)
|
|
api_module=[]
|
|
for i in res:
|
|
if i['path_name'] !='root':
|
|
data=[]
|
|
data.append(i['auth_id'])
|
|
data.append(i['path_name'])
|
|
api_module.append(data)
|
|
return schemas.Msg(code=0, msg='ok', data=api_module)
|
|
|
|
@router.post("/add_roles")
|
|
async def add_roles(
|
|
request: Request,
|
|
game:str,
|
|
data_in: schemas.Add_role,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)
|
|
) -> schemas.Msg:
|
|
"""
|
|
创建角色
|
|
"""
|
|
# try:
|
|
# res = await crud.role.add_role(db, data_in)
|
|
# return schemas.Msg(code=0, msg='ok', data=res.upserted_id)
|
|
# except Exception as e:
|
|
# return schemas.Msg(code=-1, msg='添加失败', data=str(e))
|
|
res = await crud.url_list.get_all(db)
|
|
for i in res:
|
|
if data_in.system == 1:
|
|
if data_in.name == i['name']:
|
|
return schemas.Msg(code=0, msg='该角色已存在！')
|
|
else:
|
|
if data_in.name == i['name'] and i['game'] == game:
|
|
return schemas.Msg(code=0, msg='该角色已存在！')
|
|
auth = []
|
|
if data_in.system == 1:
|
|
for i in res:
|
|
auth.append(i['auth_id'])
|
|
max_auth = 'ab' + str(int(max(auth).split('b')[-1]) + 1)
|
|
api_module = await crud.api_module.get_api_module(db)
|
|
for i in api_module:
|
|
if i['auth_id'] in data_in.path_name:
|
|
await crud.url_list.insert_url(db, schemas.Url_list(name=data_in.name, auth_id=max_auth,
|
|
path_name=i['path_name'], api_list=i['api_list'],
|
|
api_name=i['api_name'], state=i['state'],
|
|
system=data_in.system))
|
|
else:
|
|
state = []
|
|
for nu in range(len(i['state'])):
|
|
state.append(False)
|
|
if i['path_name'] != 'root':
|
|
await crud.url_list.insert_url(db, schemas.Url_list(name=data_in.name, auth_id=max_auth,
|
|
path_name=i['path_name'],
|
|
api_list=i['api_list'], api_name=i['api_name'],
|
|
state=state, system=data_in.system))
|
|
return schemas.Msg(code=0, msg='添加角色成功', data='')
|
|
else:
|
|
for i in res:
|
|
auth.append(i['auth_id'])
|
|
max_auth = 'ab' + str(int(max(auth).split('b')[-1]) + 1)
|
|
api_module = await crud.api_module.get_api_module(db)
|
|
for i in api_module:
|
|
if i['auth_id'] in data_in.path_name:
|
|
await crud.url_list.insert_urls(db, schemas.Url_lists(name=data_in.name, auth_id=max_auth,
|
|
path_name=i['path_name'], api_list=i['api_list'],
|
|
api_name=i['api_name'], state=i['state'],
|
|
system=data_in.system, game=game))
|
|
else:
|
|
state = []
|
|
for nu in range(len(i['state'])):
|
|
state.append(False)
|
|
if i['path_name'] != 'root':
|
|
await crud.url_list.insert_urls(db, schemas.Url_lists(name=data_in.name, auth_id=max_auth,
|
|
path_name=i['path_name'], game=game,
|
|
api_list=i['api_list'],
|
|
api_name=i['api_name'], state=state,
|
|
system=data_in.system))
|
|
return schemas.Msg(code=0, msg='添加角色成功', data='')
|
|
|
|
@router.get("/roles")
|
|
async def roles(
|
|
request: Request,
|
|
game: str,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)
|
|
) -> schemas.Msg:
|
|
"""
|
|
获取所有的管理员用户
|
|
"""
|
|
# res = await crud.role.dom_roles(db, game)
|
|
# return schemas.Msg(code=0, msg='ok', data=res)
|
|
res = await crud.url_list.get_all(db)
|
|
role = []
|
|
data = []
|
|
# 区分不同项目下的权限用户
|
|
for i in res:
|
|
if i['system'] == 1 and i['name'] != 'root':
|
|
role.append(i['name'])
|
|
if 'game' in i.keys():
|
|
if game == i['game']:
|
|
role.append(i['name'])
|
|
# 得到不同权限用户
|
|
role = list(set(role))
|
|
for id in role:
|
|
data_dcit = {}
|
|
data_dcit['name'] = id
|
|
auth_id = []
|
|
system = []
|
|
data_list = []
|
|
for i in res:
|
|
if i['name'] == id:
|
|
data_one = {}
|
|
auth_id.append(i['auth_id'])
|
|
system.append(i['system'])
|
|
data_one['path_name'] = i['path_name']
|
|
data_one['api_name'] = i['api_name']
|
|
data_one['api_list'] = i['api_list']
|
|
data_one['state'] = i['state']
|
|
data_list.append(data_one)
|
|
data_dcit['datalist'] = data_list
|
|
data_dcit['auth_id'] = auth_id[0]
|
|
data_dcit['system'] = system[0]
|
|
data.append(data_dcit)
|
|
return schemas.Msg(code=0, msg='ok', data=data)
|
|
|
|
@router.post("/edit_role")
|
|
async def edit_role(
|
|
request: Request,
|
|
date_in: schemas.Editname,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)
|
|
) -> schemas.Msg:
|
|
"""
|
|
修改角色名
|
|
"""
|
|
# res = await crud.role.edit_role(db, date_in)
|
|
# return schemas.Msg(code=0, msg='ok', data=res.matched_count)
|
|
await crud.url_list.edit_name(db,date_in)
|
|
return schemas.Msg(code=0,msg="ok")
|
|
|
|
@router.get("/update_api_list")
|
|
async def update_api_list(
|
|
request: Request,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user),
|
|
):
|
|
"""更新 api 列表"""
|
|
app = request.app
|
|
data = {}
|
|
for r in app.routes:
|
|
title = r.tags[0] if hasattr(r, 'description') else None
|
|
if not title:
|
|
continue
|
|
data.setdefault(title, {'list': []})
|
|
path = r.path
|
|
name = r.description if hasattr(r, 'description') else r.name
|
|
data[title]['list'].append({'api': path, 'title': name})
|
|
|
|
data = [{'title': k, 'list': v['list']} for k, v in data.items()]
|
|
for item in data:
|
|
title = item['title']
|
|
for l in item['list']:
|
|
api = l['api']
|
|
name = l['title']
|
|
add_data = schemas.UpdateApi(path=api, name=name)
|
|
await crud.api_list.update_api(db, add_data)
|
|
|
|
return schemas.Msg(code=0, msg='ok', data=1)
|
|
|
|
|
|
@router.get("/account_owner_list")
|
|
async def account_owner_list(request: Request,
|
|
game: str,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)) -> schemas.Msg:
|
|
"""获取账号owner权限"""
|
|
account_infos = await crud.user.find_many(db, {},
|
|
{'_id': False, 'name': True, 'nickname': True,
|
|
f'data_where.{game}': True})
|
|
resp = []
|
|
for account_info in account_infos:
|
|
resp.append(
|
|
{
|
|
'name': account_info.get('name'),
|
|
'nickname': account_info.get('nickname'),
|
|
'owner_list': ''
|
|
}
|
|
)
|
|
for item in account_info.get('data_where', {}).get(game, []):
|
|
if item.get('columnName') == 'owner_name':
|
|
resp[-1]['owner_list'] = ','.join(item.get('ftv', []))
|
|
break
|
|
return schemas.Msg(code=0, msg='ok', data=resp)
|
|
# @router.post("/git_owner")
|
|
# async def git_owner(request: Request,
|
|
# game: str,
|
|
# db: AsyncIOMotorDatabase = Depends(get_database),
|
|
# current_user: schemas.UserDB = Depends(deps.get_current_user)) -> schemas.Msg:
|
|
# user=await crud.user
|
|
|
|
@router.post("/update_account_owner")
|
|
async def account_owner_list(request: Request,
|
|
game: str,
|
|
data_in: schemas.OwnerList,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)) -> schemas.Msg:
|
|
"""设置账号owner权限"""
|
|
set_data = {
|
|
"columnName": "owner_name",
|
|
"tableType": "event",
|
|
"comparator": "in",
|
|
"ftv": data_in.owners
|
|
}
|
|
if not data_in.owners[0]:
|
|
res = await crud.user.update_one(db, {'name': data_in.account_name,
|
|
f'data_where.{game}': {'$exists': True}
|
|
},
|
|
{'$pull': {f'data_where.{game}': {'columnName': 'owner_name'}}}
|
|
)
|
|
return schemas.Msg(code=0, msg='ok', data=res.raw_result)
|
|
|
|
is_exists = await crud.user.find_one(db, {'name': data_in.account_name,
|
|
f'data_where.{game}': {'$exists': True},
|
|
})
|
|
if is_exists:
|
|
if await crud.user.find_one(db, {'name': data_in.account_name,
|
|
f'data_where.{game}': {'$exists': True},
|
|
f'data_where.{game}.columnName': 'owner_name'
|
|
}):
|
|
await crud.user.update_one(db, {'name': data_in.account_name,
|
|
f'data_where.{game}': {'$exists': True},
|
|
f'data_where.{game}.columnName': 'owner_name'
|
|
}, {'$set': {f'data_where.{game}.$': set_data}})
|
|
else:
|
|
await crud.user.update_one(db, {'name': data_in.account_name,
|
|
f'data_where.{game}': {'$exists': True},
|
|
}, {'$push': {f'data_where.{game}': set_data}})
|
|
else:
|
|
await crud.user.update_one(db, {'name': data_in.account_name,
|
|
}, {'$set': {f'data_where.{game}': [set_data]}})
|
|
|
|
return schemas.Msg(code=0, msg='ok')
|
|
|
|
@router.get("/all_api_board")
|
|
async def all_api_board(request: Request,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)) -> schemas.Msg:
|
|
"""显示创建项目时生成的所有api权限模板"""
|
|
res = await crud.api_board.all_api(db)
|
|
for i in res:
|
|
i['_id'] = str(i['_id'])
|
|
return schemas.Msg(code=0, msg='ok', data=res)
|
|
|
|
@router.post("/updata_api_board")
|
|
async def updata_api_board(
|
|
request: Request,
|
|
opinion: bool,
|
|
data_in: schemas.Api_board,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)) -> schemas.Msg:
|
|
"""
|
|
修改api权限模板
|
|
"""
|
|
await crud.api_board.update(db, data_in,opinion)
|
|
return schemas.Msg(code=0, msg='ok')
|
|
|
|
@router.post("/add_api_board")
|
|
async def add_api_board(
|
|
request: Request,
|
|
data_in: schemas.Api_board,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)) -> schemas.Msg:
|
|
"""
|
|
添加api权限模板
|
|
"""
|
|
res = await crud.api_board.all_api(db)
|
|
for i in res:
|
|
if data_in.name ==i['name'] and data_in.api_name == i['api_name'] and data_in.api_path == i['api_path']:
|
|
return schemas.Msg(code=-1, msg='该路径已存在')
|
|
await crud.api_board.insert(db, data_in)
|
|
return schemas.Msg(code=0, msg='ok')
|
|
|
|
@router.post("/del_api_board")
|
|
async def del_api_board(
|
|
request: Request,
|
|
data_in: schemas.Api_board,
|
|
db: AsyncIOMotorDatabase = Depends(get_database),
|
|
current_user: schemas.UserDB = Depends(deps.get_current_user)) -> schemas.Msg:
|
|
"""
|
|
删除api权限模板
|
|
"""
|
|
await crud.api_board.del_api(db, data_in)
|
|
return schemas.Msg(code=0, msg='ok') |