85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
from copy import deepcopy
|
|
|
|
import pymongo
|
|
from motor.motor_asyncio import AsyncIOMotorDatabase
|
|
|
|
from core.config import settings
|
|
from crud.base import CRUDBase
|
|
from schemas import *
|
|
from utils import *
|
|
|
|
__all__ = 'authority',
|
|
|
|
|
|
class CRUDAuthority(CRUDBase):
|
|
|
|
async def create(self, db: AsyncIOMotorDatabase, *args, **kwargs):
|
|
data = dict()
|
|
if len(args) > 0:
|
|
data['ptype'] = args[0]
|
|
if len(args) > 1:
|
|
data['v0'] = args[1]
|
|
if len(args) > 2:
|
|
data['v1'] = args[2]
|
|
if len(args) > 3:
|
|
data['v2'] = args[3]
|
|
if len(args) > 4:
|
|
data['v3'] = args[4]
|
|
if len(args) > 5:
|
|
data['v4'] = args[5]
|
|
|
|
data.update(kwargs)
|
|
await self.update_one(db, data, {'$set': data}, upsert=True)
|
|
|
|
# async def get_all_role(self, db):
|
|
# # todo 避免与用户同名
|
|
# await self.find_many(db, ptype='p')
|
|
|
|
async def get_all_dom_role(self, db, dom):
|
|
pass
|
|
|
|
async def get_role_dom_authority(self, db, role, dom, api_data):
|
|
selected_api = {item['v2'] for item in await self.find_many(db, v0=role, v1=dom)}
|
|
|
|
anonymous_api = {item['v2'] for item in await self.find_many(db, v0='anonymous')}
|
|
|
|
api_data = deepcopy(api_data)
|
|
|
|
for api, data in api_data.items():
|
|
if api in selected_api or '*' in selected_api or api in anonymous_api:
|
|
data['selected'] = True
|
|
else:
|
|
data['selected'] = False
|
|
res = {}
|
|
for api, item in api_data.items():
|
|
res.setdefault(item['title'], list())
|
|
res[item['title']].append(item)
|
|
return res
|
|
|
|
async def set_data_auth(self, db: AsyncIOMotorDatabase, data_in, game):
|
|
v0 = data_in.username
|
|
v2 = game
|
|
data_auth_id = data_in.data_auth_id
|
|
await self.update_one(db, {'ptype': 'g', 'v0': v0, 'v2': v2}, {'$set': {'data_auth_id': data_auth_id}},
|
|
upsert=True)
|
|
|
|
async def get_data_auth(self, db, username, game):
|
|
v0 = username
|
|
v2 = game
|
|
res = await self.find_one(db, {'ptype': 'g', 'v0': v0, 'v2': v2, 'data_auth_id': {'$exists': 1}},
|
|
{'_id': 0, 'data_auth_id': 1})
|
|
# 没有设置或者设置为*认为是全部事件
|
|
return res.get('data_auth_id') if res.get('data_auth_id', '*') != '*' else None
|
|
|
|
async def get_all_user(self, db: AsyncIOMotorDatabase):
|
|
return await self.distinct(db, 'v0', {'ptype': 'g'})
|
|
|
|
async def create_index(self, db: AsyncIOMotorDatabase):
|
|
await db[self.coll_name].create_index(
|
|
[('ptype', pymongo.DESCENDING), ('v0', pymongo.DESCENDING), ('v1', pymongo.DESCENDING),
|
|
('v2', pymongo.DESCENDING), ('v3', pymongo.DESCENDING)],
|
|
unique=True)
|
|
|
|
|
|
authority = CRUDAuthority(settings.CASBIN_COLL)
|