162 lines
7.9 KiB
Python
162 lines
7.9 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from role.models import PermissionTags, Permissions, Roles
|
|
from role.serializers import serialize_permission, serialize_permission_tag, serialize_role
|
|
|
|
|
|
class Role:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def _bundle_payload(self, permission: Permissions) -> dict:
|
|
tag_ids = [int(i) for i in (permission.permission_tags or [])]
|
|
tags = await PermissionTags.get_permission_tags_by_ids(self.session, tag_ids)
|
|
by_id = {t.id: t.tag_name for t in tags}
|
|
tag_names = [by_id[i] for i in tag_ids if i in by_id]
|
|
return serialize_permission(permission, tag_names=tag_names)
|
|
|
|
async def _role_payload(self, role: Roles) -> dict:
|
|
perm_ids = [int(i) for i in (role.permissions or [])]
|
|
bundles_orm = await Permissions.get_permissions_by_ids(self.session, perm_ids)
|
|
by_id = {b.id: b for b in bundles_orm}
|
|
bundles = []
|
|
for pid in perm_ids:
|
|
bundle = by_id.get(pid)
|
|
if bundle is not None:
|
|
bundles.append(await self._bundle_payload(bundle))
|
|
tags = await Roles.resolve_tags(self.session, role)
|
|
return serialize_role(role, bundles=bundles, permissions=list(tags))
|
|
|
|
async def get_roles(self, top, skip, search=None):
|
|
rows = await Roles.get_roles(self.session, top, skip, search)
|
|
return [await self._role_payload(r) for r in rows]
|
|
|
|
async def get_role_by_id(self, record_id):
|
|
role = await Roles.get_role_by_id(self.session, int(record_id))
|
|
if not role or role.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
return await self._role_payload(role)
|
|
|
|
async def count_roles(self, search=None):
|
|
return await Roles.count_roles(self.session, search)
|
|
|
|
async def create_role(self, payload):
|
|
name = (payload.get("role_name") or "").strip()
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="role_name is required")
|
|
if await Roles.get_role_by_name(self.session, name):
|
|
raise HTTPException(status_code=409, detail="Role name already exists")
|
|
fields = {
|
|
"role_name": name,
|
|
"description": payload.get("description"),
|
|
"permissions": list(payload.get("permissions") or []),
|
|
"is_system": False,
|
|
"is_active": payload.get("is_active", True),
|
|
"is_deleted": False,
|
|
}
|
|
role = await Roles.insert_role(self.session, fields)
|
|
return await self._role_payload(role)
|
|
|
|
async def update_role(self, record_id, payload):
|
|
role = await Roles.get_role_by_id(self.session, int(record_id))
|
|
if not role or role.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
fields = {}
|
|
if "role_name" in payload and payload["role_name"] is not None:
|
|
new_name = payload["role_name"].strip()
|
|
if role.is_system and new_name != role.role_name:
|
|
raise HTTPException(status_code=409, detail="System roles cannot be renamed")
|
|
if new_name != role.role_name:
|
|
clash = await Roles.get_role_by_name(self.session, new_name)
|
|
if clash:
|
|
raise HTTPException(status_code=409, detail="Role name already exists")
|
|
fields["role_name"] = new_name
|
|
if "description" in payload and payload["description"] is not None:
|
|
fields["description"] = payload["description"]
|
|
if "permissions" in payload and payload["permissions"] is not None:
|
|
fields["permissions"] = list(payload["permissions"])
|
|
if "is_active" in payload and payload["is_active"] is not None:
|
|
fields["is_active"] = payload["is_active"]
|
|
updated = await Roles.update_role(self.session, int(record_id), fields)
|
|
return await self._role_payload(updated)
|
|
|
|
async def delete_role(self, record_id):
|
|
role = await Roles.get_role_by_id(self.session, int(record_id))
|
|
if not role or role.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
if role.is_system:
|
|
raise HTTPException(status_code=409, detail="System roles cannot be deleted")
|
|
deleted = await Roles.soft_delete_role(self.session, int(record_id))
|
|
return await self._role_payload(deleted)
|
|
|
|
async def get_permissions(self, top, skip, search=None):
|
|
rows = await Permissions.get_permissions(self.session, top, skip, search)
|
|
return [await self._bundle_payload(r) for r in rows]
|
|
|
|
async def get_permission_by_id(self, record_id):
|
|
row = await Permissions.get_permission_by_id(self.session, int(record_id))
|
|
if not row or row.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Permission bundle not found")
|
|
return await self._bundle_payload(row)
|
|
|
|
async def count_permissions(self, search=None):
|
|
return await Permissions.count_permissions(self.session, search)
|
|
|
|
async def create_permission(self, payload):
|
|
name = (payload.get("name") or "").strip()
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="name is required")
|
|
if await Permissions.get_permission_by_name(self.session, name):
|
|
raise HTTPException(status_code=409, detail="Permission bundle name already exists")
|
|
fields = {
|
|
"name": name,
|
|
"description": payload.get("description"),
|
|
"permission_tags": list(payload.get("permission_tags") or []),
|
|
"is_system": False,
|
|
"is_active": payload.get("is_active", True),
|
|
"is_deleted": False,
|
|
}
|
|
row = await Permissions.insert_permission(self.session, fields)
|
|
return await self._bundle_payload(row)
|
|
|
|
async def update_permission(self, record_id, payload):
|
|
row = await Permissions.get_permission_by_id(self.session, int(record_id))
|
|
if not row or row.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Permission bundle not found")
|
|
fields = {}
|
|
if "name" in payload and payload["name"] is not None:
|
|
new_name = payload["name"].strip()
|
|
if row.is_system and new_name != row.name:
|
|
raise HTTPException(
|
|
status_code=409, detail="System permission bundles cannot be renamed"
|
|
)
|
|
if new_name != row.name:
|
|
clash = await Permissions.get_permission_by_name(self.session, new_name)
|
|
if clash:
|
|
raise HTTPException(
|
|
status_code=409, detail="Permission bundle name already exists"
|
|
)
|
|
fields["name"] = new_name
|
|
if "description" in payload and payload["description"] is not None:
|
|
fields["description"] = payload["description"]
|
|
if "permission_tags" in payload and payload["permission_tags"] is not None:
|
|
fields["permission_tags"] = list(payload["permission_tags"])
|
|
if "is_active" in payload and payload["is_active"] is not None:
|
|
fields["is_active"] = payload["is_active"]
|
|
updated = await Permissions.update_permission(self.session, int(record_id), fields)
|
|
return await self._bundle_payload(updated)
|
|
|
|
async def get_permission_tags(self, top, skip, search=None):
|
|
rows = await PermissionTags.get_permission_tags(self.session, top, skip, search)
|
|
return [serialize_permission_tag(r) for r in rows]
|
|
|
|
async def get_permission_tag_by_id(self, record_id):
|
|
row = await PermissionTags.get_permission_tag_by_id(self.session, int(record_id))
|
|
if not row or row.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Permission tag not found")
|
|
return serialize_permission_tag(row)
|
|
|
|
async def count_permission_tags(self, search=None):
|
|
return await PermissionTags.count_permission_tags(self.session, search)
|