HR-ATS-Portal/backend/role/app.py

from fastapi import APIRouter, Depends, Query
from fastapi.responses import JSONResponse
from fastapi import HTTPException
from db_setup import get_session
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel
from role.views import Role
from users.permissions import PermissionTag, require_permission
from dotenv import load_dotenv
load_dotenv()

router = APIRouter()


class RoleCreate(BaseModel):
    role_name: str
    description: str | None = None
    permissions: list[int] | None = None
    is_active: bool = True


class RoleUpdate(BaseModel):
    role_name: str | None = None
    description: str | None = None
    permissions: list[int] | None = None
    is_active: bool | None = None


class PermissionCreate(BaseModel):
    name: str
    description: str | None = None
    permission_tags: list[int] | None = None
    is_active: bool = True


class PermissionUpdate(BaseModel):
    name: str | None = None
    description: str | None = None
    permission_tags: list[int] | None = None
    is_active: bool | None = None


@router.get("/roles/fetch")
async def fetch_roles(
    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_VIEW)),
    record_id: int | None = Query(None),
    search: str | None = Query(None),
    top: int | None = Query(None),
    skip: int = Query(0, ge=0),
    session: AsyncSession = Depends(get_session),
):
    try:
        service=Role(session=session)
        if record_id is not None:
            item=await service.get_role_by_id(record_id)
            return JSONResponse(content={"data":item,"total":1,"status_code":200})
        items=await service.get_roles(top,skip,search)
        total=await service.count_roles(search)
        return JSONResponse(content={"data":items,"total":total,"status_code":200})
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500,detail=str(e))


@router.post("/roles/create")
async def create_role(
    payload: RoleCreate,
    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_CREATE)),
    session: AsyncSession = Depends(get_session),
):
    try:
        service=Role(session=session)
        data=await service.create_role(payload.model_dump())
        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("/roles/update")
async def update_role(
    payload: RoleUpdate,
    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_EDIT)),
    record_id: int = Query(...),
    session: AsyncSession = Depends(get_session),
):
    try:
        service=Role(session=session)
        data=await service.update_role(record_id,payload.model_dump())
        return JSONResponse(content={"data":data,"status_code":200})
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500,detail=str(e))


@router.delete("/roles/delete")
async def delete_role(
    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_DELETE)),
    record_id: int = Query(...),
    session: AsyncSession = Depends(get_session),
):
    try:
        service=Role(session=session)
        data=await service.delete_role(record_id)
        return JSONResponse(content={"data":data,"status_code":200})
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500,detail=str(e))


@router.get("/permissions/fetch")
async def fetch_permissions(
    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_VIEW)),
    record_id: int | None = Query(None),
    search: str | None = Query(None),
    top: int | None = Query(None),
    skip: int = Query(0, ge=0),
    session: AsyncSession = Depends(get_session),
):
    try:
        service=Role(session=session)
        if record_id is not None:
            item=await service.get_permission_by_id(record_id)
            return JSONResponse(content={"data":item,"total":1,"status_code":200})
        items=await service.get_permissions(top,skip,search)
        total=await service.count_permissions(search)
        return JSONResponse(content={"data":items,"total":total,"status_code":200})
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500,detail=str(e))


@router.post("/permissions/create")
async def create_permission(
    payload: PermissionCreate,
    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_MANAGE)),
    session: AsyncSession = Depends(get_session),
):
    try:
        service=Role(session=session)
        data=await service.create_permission(payload.model_dump())
        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("/permissions/update")
async def update_permission(
    payload: PermissionUpdate,
    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_MANAGE)),
    record_id: int = Query(...),
    session: AsyncSession = Depends(get_session),
):
    try:
        service=Role(session=session)
        data=await service.update_permission(record_id,payload.model_dump())
        return JSONResponse(content={"data":data,"status_code":200})
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500,detail=str(e))


@router.get("/permission-tags/fetch")
async def fetch_permission_tags(
    current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_VIEW)),
    record_id: int | None = Query(None),
    search: str | None = Query(None),
    top: int | None = Query(None),
    skip: int = Query(0, ge=0),
    session: AsyncSession = Depends(get_session),
):
    try:
        service=Role(session=session)
        if record_id is not None:
            item=await service.get_permission_tag_by_id(record_id)
            return JSONResponse(content={"data":item,"total":1,"status_code":200})
        items=await service.get_permission_tags(top,skip,search)
        total=await service.count_permission_tags(search)
        return JSONResponse(content={"data":items,"total":total,"status_code":200})
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500,detail=str(e))