asssignment role api

backend/users/app.py

@@ -26,10 +26,13 @@ class UserUpdate(BaseModel):
     name: str | None = None
     email: EmailStr | None = None
     password: str | None = None
-    role_id: int | None = None
     is_active: bool | None = None
 
 
+class RoleAssign(BaseModel):
+    role_id: int
+
+
 class UserLogin(BaseModel):
     password: str
     email: EmailStr | None = None
@@ -90,7 +93,7 @@ async def create_user(
 ):
     try:
         service=User(session=session)
-        user=await service.create_user(payload.model_dump())
+        user=await service.create_user(payload.model_dump(),current_user)
         tokens=serialize_token(create_access_token(user),create_refresh_token(user),user)
         return JSONResponse(content={**tokens,"status_code":200})
     except HTTPException:
@@ -132,7 +135,40 @@ async def update_user(
 ):
     try:
         service=User(session=session)
-        data=await service.update_user(record_id,payload.model_dump())
+        data=await service.update_user(record_id,payload.model_dump(exclude_unset=True))
+        return JSONResponse(content={"data":data,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
+@router.put("/users/assign-role")
+async def assign_role(
+    payload: RoleAssign,
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_EDIT)),
+    record_id: str = Query(...),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=User(session=session)
+        data=await service.assign_role(record_id,payload.role_id,current_user)
+        return JSONResponse(content={"data":data,"status_code":200})
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=500,detail=str(e))
+
+
+@router.put("/users/remove-role")
+async def remove_role(
+    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_EDIT)),
+    record_id: str = Query(...),
+    session: AsyncSession = Depends(get_session),
+):
+    try:
+        service=User(session=session)
+        data=await service.remove_role(record_id,current_user)
         return JSONResponse(content={"data":data,"status_code":200})
     except HTTPException:
         raise

backend/users/views.py

@@ -1,5 +1,7 @@
 from fastapi import HTTPException
+from role.models import Roles
 from users.models import Users
+from users.permissions import PermissionTag,has_permission
 from users.serializers import serialize_user
 from users.plugins import clean_user_payload,verify_password,decode_token
 from dotenv import load_dotenv
@@ -7,14 +9,33 @@ load_dotenv()
 from sqlalchemy.ext.asyncio import AsyncSession
 import jwt
 
+
 class User:
     def __init__(self,session:AsyncSession):
         self.session=session
 
-    async def create_user(self,payload):
+    async def _check_role_assignment(self,current_user,role_id,existing_role_id=None):
+        if role_id==existing_role_id:
+            return
+        if not has_permission(current_user.get("permissions") or [],PermissionTag.RBAC_USERS_MANAGE):
+            raise HTTPException(status_code=403,detail="Assigning a role requires rbac_users.manage")
+        if role_id is None:
+            return
+        role=await Roles.get_role_by_id(self.session,role_id)
+        if role is None or role.is_deleted:
+            raise HTTPException(status_code=404,detail="Role not found")
+        if not role.is_active:
+            raise HTTPException(status_code=400,detail="Role is not active")
+        target=set(await Roles.resolve_tags(self.session,role))
+        missing=sorted(target-set(current_user.get("permissions") or []))
+        if missing:
+            raise HTTPException(status_code=403,detail=f"Cannot assign a role with permissions you do not hold: {', '.join(missing)}")
+
+    async def create_user(self,payload,current_user):
         existing=await Users.get_user_by_email(self.session,payload.get("email"))
         if existing:
             raise HTTPException(status_code=409,detail="Email already registered")
+        await self._check_role_assignment(current_user,payload.get("role_id"),None)
         # this is for password hasshing
         fields=clean_user_payload(payload)
         if not fields.get("password"):
@@ -44,6 +65,22 @@ class User:
         updated=await Users.update_user(self.session,record_id,fields)
         return serialize_user(updated)
 
+    async def assign_role(self,record_id,role_id,current_user):
+        user=await Users.get_user_by_id(self.session,record_id)
+        if not user:
+            raise HTTPException(status_code=404,detail="User not found")
+        await self._check_role_assignment(current_user,role_id,user.role_id)
+        updated=await Users.update_user(self.session,record_id,{"role_id":role_id})
+        return serialize_user(updated)
+
+    async def remove_role(self,record_id,current_user):
+        user=await Users.get_user_by_id(self.session,record_id)
+        if not user:
+            raise HTTPException(status_code=404,detail="User not found")
+        await self._check_role_assignment(current_user,None,user.role_id)
+        updated=await Users.update_user(self.session,record_id,{"role_id":None})
+        return serialize_user(updated)
+
     async def delete_user(self,record_id):
         user=await Users.soft_delete_user(self.session,record_id)
         if not user: