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 set_role_matrix(self, record_id, permission_tags): """Write the Access Control grid onto one overlay bundle for this role. Shared system bundles are not mutated. The role then points at that overlay only, so a ticked cell is the grant and an unticked cell is not. """ 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") tag_ids = sorted({int(i) for i in (permission_tags or [])}) found = await PermissionTags.get_permission_tags_by_ids(self.session, tag_ids) unknown = sorted(set(tag_ids) - {t.id for t in found}) if unknown: raise HTTPException( status_code=422, detail=f"Unknown or inactive permission tag ids: {unknown}", ) overlay_name = f"role_{role.id}_matrix" bundle = await Permissions.get_permission_by_name(self.session, overlay_name) if bundle is None: bundle = await Permissions.insert_permission( self.session, { "name": overlay_name, "description": f"Access Control matrix for {role.role_name}", "permission_tags": tag_ids, "is_system": False, "is_active": True, "is_deleted": False, }, ) else: await Permissions.update_permission( self.session, int(bundle.id), {"permission_tags": tag_ids}, ) updated = await Roles.update_role( self.session, int(record_id), {"permissions": [int(bundle.id)]}, ) 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) class PermissionBundle: """Sets the exact permission-tag set on one bundle, for the permission matrix. `id` is a permissions.id from /permissions/fetch. Every role holding that bundle picks the change up on its next request, because Roles.resolve_tags runs live. """ def __init__(self,session:AsyncSession): self.session=session async def set_role_tags(self,payload): bundle=await Permissions.get_permission_by_id(self.session,int(payload.get("id"))) if not bundle or bundle.is_deleted: raise HTTPException(status_code=404,detail="Permission bundle not found") tag_ids=sorted(set(payload.get("permission_tags") or [])) # an unknown or inactive id resolves to nothing, so the box would silently # refuse to stay ticked. reject instead of granting less than was asked. found=await PermissionTags.get_permission_tags_by_ids(self.session,tag_ids) unknown=sorted(set(tag_ids)-{t.id for t in found}) if unknown: raise HTTPException(status_code=422,detail=f"Unknown or inactive permission tag ids: {unknown}") # build fields explicitly: update_permission setattr's whatever it is given, so # passing the payload through would write name=None and hit the NOT NULL. fields={"permission_tags":tag_ids} if payload.get("name") is not None: name=payload["name"].strip() if bundle.is_system and name!=bundle.name: raise HTTPException(status_code=409,detail="System permission bundles cannot be renamed") clash=await Permissions.get_permission_by_name(self.session,name) if clash and clash.id!=bundle.id: raise HTTPException(status_code=409,detail="Permission bundle name already exists") fields["name"]=name if payload.get("description") is not None: fields["description"]=payload["description"] if payload.get("is_active") is not None: fields["is_active"]=payload["is_active"] await Permissions.update_permission(self.session,int(bundle.id),fields) return await Role(session=self.session).get_permission_by_id(int(bundle.id))