66 lines
2.4 KiB
Python
66 lines
2.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from app.core.db import get_db
|
|
from app.core.deps import get_current_user
|
|
from app.schemas.permission import PermissionOut, PermissionCreate, PermissionUpdate
|
|
from app.models import Permission, RolePermission, UserRole, User
|
|
from app.models.enums import StatusEnum
|
|
|
|
|
|
router = APIRouter(prefix="/permissions", tags=["permissions"])
|
|
|
|
|
|
@router.get("", response_model=list[PermissionOut])
|
|
def list_permissions(db: Session = Depends(get_db)):
|
|
return db.query(Permission).order_by(Permission.level, Permission.sort_order).all()
|
|
|
|
|
|
@router.post("", response_model=PermissionOut)
|
|
def create_permission(payload: PermissionCreate, db: Session = Depends(get_db)):
|
|
exists = db.query(Permission).filter(Permission.code == payload.code).first()
|
|
if exists:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Permission code exists")
|
|
item = Permission(**payload.model_dump())
|
|
db.add(item)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
@router.put("/{perm_id}", response_model=PermissionOut)
|
|
def update_permission(perm_id: int, payload: PermissionUpdate, db: Session = Depends(get_db)):
|
|
item = db.query(Permission).filter(Permission.perm_id == perm_id).first()
|
|
if not item:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Permission not found")
|
|
for k, v in payload.model_dump(exclude_unset=True).items():
|
|
setattr(item, k, v)
|
|
db.commit()
|
|
db.refresh(item)
|
|
return item
|
|
|
|
|
|
@router.delete("/{perm_id}")
|
|
def delete_permission(perm_id: int, db: Session = Depends(get_db)):
|
|
item = db.query(Permission).filter(Permission.perm_id == perm_id).first()
|
|
if not item:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Permission not found")
|
|
db.delete(item)
|
|
db.commit()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.get("/me")
|
|
def my_permissions(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
rows = (
|
|
db.query(Permission.code)
|
|
.join(RolePermission, RolePermission.perm_id == Permission.perm_id)
|
|
.join(UserRole, UserRole.role_id == RolePermission.role_id)
|
|
.filter(UserRole.user_id == current_user.user_id)
|
|
.filter(Permission.status == int(StatusEnum.ENABLED))
|
|
.all()
|
|
)
|
|
return [r[0] for r in rows]
|