235 lines
7.9 KiB
Python
235 lines
7.9 KiB
Python
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,PermissionBundle
|
|
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
|
|
|
|
class RolePermissionTagsUpdate(BaseModel):
|
|
"""`id` is a permissions.id from /permissions/fetch; permission_tags is the exact set."""
|
|
id: int
|
|
name: str | None = None
|
|
description: str | None = None
|
|
permission_tags: list[int]
|
|
is_active: bool | None = None
|
|
|
|
|
|
class RoleMatrixUpdate(BaseModel):
|
|
"""Exact tag ids for one role. Saved onto that role's overlay bundle."""
|
|
permission_tags: list[int]
|
|
|
|
|
|
@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.put("/roles/matrix/update")
|
|
async def update_role_matrix(
|
|
payload: RoleMatrixUpdate,
|
|
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.set_role_matrix(record_id,payload.permission_tags)
|
|
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/permission-tags/update")
|
|
async def update_role_permission_tags(
|
|
payload: RolePermissionTagsUpdate,
|
|
current_user: dict = Depends(require_permission(PermissionTag.RBAC_USERS_MANAGE)),
|
|
session: AsyncSession = Depends(get_session),
|
|
):
|
|
try:
|
|
service=PermissionBundle(session=session)
|
|
data=await service.set_role_tags(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))
|