from database import db from models import Organization, User from sqlalchemy.orm.attributes import flag_modified from utils import Perm class RoleService: @staticmethod def create_role(org: Organization, role: str, perms: list[Perm]) -> dict: if role in org.roles: raise ValueError(f"Role {role} already exists in organization {org.name}") roles = org.roles.copy() roles[role] = { "permissions": Perm.get_int(perms), "users": [], "status": "active" } org.roles = roles flag_modified(org, "roles") db.commit() db.refresh(org) return org.roles[role] @staticmethod def delete_role(org: Organization, role: str) -> dict: roles = org.roles.copy() del roles[role] org.roles = roles flag_modified(org, "roles") db.commit() db.refresh(org) return roles @staticmethod def activate_role(org: Organization, role: str) -> Organization: if role not in org.roles: raise ValueError(f"Role {role} does not exist in organization {org.name}") if org.roles[role]["status"] == "active": raise ValueError(f"Role {role} is already active in organization {org.name}") roles = org.roles.copy() roles[role]["status"] = "active" org.roles = roles flag_modified(org, "roles") db.commit() db.refresh(org) return org @staticmethod def suspend_role(org: Organization, role: str) -> Organization: if role == "manager": raise ValueError(f"Role {role} cannot be suspended in organization {org.name}") if role not in org.roles: raise ValueError(f"Role {role} does not exist in organization {org.name}") if org.roles[role]["status"] == "suspended": raise ValueError(f"Role {role} is already suspended in organization {org.name}") roles = org.roles.copy() roles[role]["status"] = "suspended" org.roles = roles flag_modified(org, "roles") db.commit() db.refresh(org) return org @staticmethod def get_role(org: Organization, role: str) -> dict: if role not in org.roles: raise ValueError(f"Role {role} does not exist in organization {org.name}") return org.roles[role] @staticmethod def check_role_permission(org: Organization, role: str, perm: Perm) -> bool: role_perms = org.roles[role]["permissions"] return Perm.check_perm(role_perms, perm.value) @staticmethod def check_user_permission(user: User, org: Organization, perm: Perm) -> bool: for role in user.roles[org.id]: if RoleService.check_role_permission(org, role, perm): return True return False @staticmethod def list_roles(org: Organization) -> dict: return org.roles @staticmethod def list_users_in_role(org: Organization, role: str) -> list: return org.roles[role]["users"] @staticmethod def list_roles_for_user(user: User, org: Organization) -> list: return [role for role in org.roles if user.id in org.roles[role]["users"]] @staticmethod def list_perms_for_role(org: Organization, role: str, return_str=False) -> list[Perm | str]: return Perm.get_perms(org.roles[role]["permissions"], return_str) @staticmethod def list_roles_for_perm(org: Organization, perm: Perm) -> list: roles = [] for role in org.roles: if RoleService.check_role_permission(org, role, perm): roles.append(role) return roles @staticmethod def add_role_to_user(user: User, org: Organization, role: str) -> User: if role not in org.roles: raise ValueError(f"Role {role} does not exist in organization {org.name}") if user.id in org.roles[role]["users"]: raise ValueError(f"User {user.username} already has role {role} in organization {org.name}") if org.roles[role]["status"] != "active": raise ValueError(f"Role {role} is not active in organization {org.name}") roles = user.roles.copy() roles[org.id] = role user.roles = roles flag_modified(user, "roles") db.commit() db.refresh(user) roles = org.roles.copy() roles[role]["users"].append(user.id) org.roles = roles flag_modified(org, "roles") db.commit() db.refresh(org) return user @staticmethod def remove_role_from_user(user: User, org: Organization, role: str) -> User: if role not in org.roles: raise ValueError(f"Role {role} does not exist in organization {org.name}") if user.id not in org.roles[role]["users"]: raise ValueError(f"User {user.username} does not have role {role} in organization {org.name}") if org.roles[role]["status"] != "active": raise ValueError(f"Role {role} is not active in organization {org.name}") roles = user.roles.copy() roles.pop(org.id) user.roles = roles flag_modified(user, "roles") db.commit() db.refresh(user) roles = org.roles.copy() roles[role]["users"].remove(user.id) org.roles = roles flag_modified(org, "roles") db.commit() db.refresh(org) return user @staticmethod def add_perm_to_role(org: Organization, role: str, perm: Perm) -> dict: if role not in org.roles: raise ValueError(f"Role {role} does not exist in organization {org.name}") if org.roles[role]["status"] != "active": raise ValueError(f"Role {role} is not active in organization {org.name}") roles = org.roles.copy() roles[role]["permissions"] |= perm.value org.roles = roles flag_modified(org, "roles") db.commit() db.refresh(org) return org.roles[role] @staticmethod def remove_perm_from_role(org: Organization, role: str, perm: Perm) -> dict: if role not in org.roles: raise ValueError(f"Role {role} does not exist in organization {org.name}") if org.roles[role]["status"] != "active": raise ValueError(f"Role {role} is not active in organization {org.name}") roles = org.roles.copy() roles[role]["permissions"] &= ~perm.value org.roles = roles flag_modified(org, "roles") db.commit() db.refresh(org) return org.roles[role]