This commit is contained in:
@ -29,10 +29,21 @@ async def _to_entity_chain(el: _EntityLike):
|
|||||||
|
|
||||||
|
|
||||||
class PermManager:
|
class PermManager:
|
||||||
|
"""
|
||||||
|
权限管理模块
|
||||||
|
"""
|
||||||
|
|
||||||
def __init__(self, db: DatabaseManager) -> None:
|
def __init__(self, db: DatabaseManager) -> None:
|
||||||
self.db = db
|
self.db = db
|
||||||
|
|
||||||
async def get_permission_info(self, entities: _EntityLike, key: str):
|
async def get_permission_info(
|
||||||
|
self, entities: _EntityLike, key: str
|
||||||
|
) -> tuple[PermEntity, str, bool] | None:
|
||||||
|
"""
|
||||||
|
获得一个权限实体或权限实体串对一个 key 的权限信息。若未入库(默认值)则
|
||||||
|
代表没有该权限相关的记录
|
||||||
|
"""
|
||||||
|
|
||||||
entities = await _to_entity_chain(entities)
|
entities = await _to_entity_chain(entities)
|
||||||
key = key.removesuffix("*").removesuffix(".")
|
key = key.removesuffix("*").removesuffix(".")
|
||||||
key_split = key.split(".")
|
key_split = key.split(".")
|
||||||
@ -52,17 +63,29 @@ class PermManager:
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
async def check_has_permission(self, entities: _EntityLike, key: str) -> bool:
|
async def check_has_permission(self, entities: _EntityLike, key: str) -> bool:
|
||||||
|
"""
|
||||||
|
检查一个权限实体或者权限实体串是否有权限
|
||||||
|
"""
|
||||||
|
|
||||||
res = await self.get_permission_info(entities, key)
|
res = await self.get_permission_info(entities, key)
|
||||||
if res is None:
|
if res is None:
|
||||||
return False
|
return False
|
||||||
return res[2]
|
return res[2]
|
||||||
|
|
||||||
async def update_permission(self, entity: PermEntity, key: str, perm: bool | None):
|
async def update_permission(self, entity: PermEntity, key: str, perm: bool | None):
|
||||||
|
"""
|
||||||
|
更新一个具体的权限实体的一则权限
|
||||||
|
"""
|
||||||
|
|
||||||
async with self.db.get_conn() as conn:
|
async with self.db.get_conn() as conn:
|
||||||
repo = PermRepo(conn)
|
repo = PermRepo(conn)
|
||||||
await repo.update_perm_info(entity, key, perm)
|
await repo.update_perm_info(entity, key, perm)
|
||||||
|
|
||||||
async def list_permission(self, entities: _EntityLike, query: PagerQuery):
|
async def list_permission(self, entities: _EntityLike, query: PagerQuery):
|
||||||
|
"""
|
||||||
|
列出一个权限实体或权限实体串拥有的所有权限记录
|
||||||
|
"""
|
||||||
|
|
||||||
entities = await _to_entity_chain(entities)
|
entities = await _to_entity_chain(entities)
|
||||||
async with self.db.get_conn() as conn:
|
async with self.db.get_conn() as conn:
|
||||||
repo = PermRepo(conn)
|
repo = PermRepo(conn)
|
||||||
|
|||||||
@ -22,6 +22,11 @@ class PermEntity:
|
|||||||
|
|
||||||
|
|
||||||
def get_entity_chain_of_entity(entity: PermEntity) -> list[PermEntity]:
|
def get_entity_chain_of_entity(entity: PermEntity) -> list[PermEntity]:
|
||||||
|
"""
|
||||||
|
获得一个权限实体的权限串。实际上返回三个权限,从小到大分别是用户、平台全体和
|
||||||
|
系统全局的权限实体。
|
||||||
|
"""
|
||||||
|
|
||||||
return [
|
return [
|
||||||
PermEntity("sys", "global", "global"),
|
PermEntity("sys", "global", "global"),
|
||||||
PermEntity(entity.platform, "global", "global"),
|
PermEntity(entity.platform, "global", "global"),
|
||||||
@ -30,6 +35,10 @@ def get_entity_chain_of_entity(entity: PermEntity) -> list[PermEntity]:
|
|||||||
|
|
||||||
|
|
||||||
async def get_entity_chain(event: Event) -> list[PermEntity]: # pragma: no cover
|
async def get_entity_chain(event: Event) -> list[PermEntity]: # pragma: no cover
|
||||||
|
"""
|
||||||
|
获得一个 Nonebot Event 的权限实体串。
|
||||||
|
"""
|
||||||
|
|
||||||
entities = [PermEntity("sys", "global", "global")]
|
entities = [PermEntity("sys", "global", "global")]
|
||||||
|
|
||||||
if isinstance(event, OB11Event):
|
if isinstance(event, OB11Event):
|
||||||
|
|||||||
Reference in New Issue
Block a user