180 lines
8.1 KiB
Python
180 lines
8.1 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import select, func
|
|
from typing import List
|
|
import ulid
|
|
|
|
from app.core.database.db_session import get_db
|
|
from app.core.permissions.RoleChecker import get_current_user, RoleChecker
|
|
from app.models.UserModel import User
|
|
from app.models.RoleModel import Role
|
|
from app.models.PermissionModel import Permission, RolePermission
|
|
from app.schemas.RolePermission import RoleResponse, RoleCreate, PermissionResponse, RolePermissionUpdate
|
|
|
|
router = APIRouter(prefix="/api/v1/role-permissions", tags=["Roles & Permissions Control"])
|
|
|
|
@router.get("/roles/all", response_model=List[RoleResponse])
|
|
def get_all_roles(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin"]))
|
|
):
|
|
stmt = select(Role).where(Role.deleted_at.is_(None))
|
|
roles = db.execute(stmt).scalars().all()
|
|
return roles
|
|
|
|
@router.post("/roles/create", response_model=RoleResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_role(
|
|
data: RoleCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin"]))
|
|
):
|
|
existing = db.execute(select(Role).where(Role.role_name == data.role_name)).scalar_one_or_none()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="Role name already exists")
|
|
|
|
role_id = str(ulid.ULID())
|
|
new_role = Role(
|
|
role_id=role_id,
|
|
role_name=data.role_name,
|
|
role_prefix=data.role_prefix,
|
|
description=data.description,
|
|
is_system=False,
|
|
is_active=True
|
|
)
|
|
db.add(new_role)
|
|
db.commit()
|
|
db.refresh(new_role)
|
|
return new_role
|
|
|
|
@router.delete("/roles/delete/{role_id}", status_code=status.HTTP_200_OK)
|
|
def delete_role(
|
|
role_id: str,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin"]))
|
|
):
|
|
role = db.execute(select(Role).where(Role.role_id == role_id)).scalar_one_or_none()
|
|
if not role:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
if role.is_system:
|
|
raise HTTPException(status_code=400, detail="System roles cannot be deleted")
|
|
|
|
user_assigned = db.execute(select(User).where(User.role_id == role_id, User.deleted_at.is_(None))).scalars().all()
|
|
if len(user_assigned) > 0:
|
|
raise HTTPException(status_code=400, detail="Cannot delete role because it is currently assigned to users")
|
|
|
|
role.deleted_at = db.execute(select(func.now())).scalar()
|
|
db.commit()
|
|
return {"detail": "Role deleted successfully"}
|
|
|
|
@router.get("/permissions/all", response_model=List[PermissionResponse])
|
|
def get_all_permissions(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin"]))
|
|
):
|
|
stmt = select(Permission).where(Permission.is_active.is_(True))
|
|
existing_permissions = db.execute(stmt).scalars().all()
|
|
existing_codes = {p.permission_code for p in existing_permissions}
|
|
|
|
seed_data = [
|
|
# User Management
|
|
("users:read", "users", "general", "View user lists and profile details"),
|
|
("users:write", "users", "general", "Create and modify staff user accounts"),
|
|
("users:delete", "users", "general", "Delete or deactivate staff user accounts"),
|
|
("roles:manage", "roles", "general", "Create, edit and assign roles & access control policies"),
|
|
("customers:manage", "customers", "general", "Manage registered e-commerce customers and addresses"),
|
|
("staff:manage", "staff", "general", "Manage staff directory, departments, and designations"),
|
|
# Catalog Management
|
|
("catalog:read", "catalog", "general", "View categories, brands, devices, and sales products"),
|
|
("catalog:write", "catalog", "general", "Create and edit product catalog items"),
|
|
("catalog:delete", "catalog", "general", "Delete catalog items, brands, and categories"),
|
|
# Marketing & Storefront
|
|
("marketing:manage", "marketing", "general", "Manage homepage sections, banners, and layout controls"),
|
|
# Inventory & Purchases
|
|
("inventory:manage", "inventory", "general", "Manage stock movements, warehouses, and purchase orders"),
|
|
# Sales & Billing
|
|
("sales:manage", "sales", "general", "Manage sales orders, invoices, and POS synchronization"),
|
|
("orders:read", "orders", "general", "View customer sales orders and fulfillment status"),
|
|
("orders:write", "orders", "general", "Update order status, shipping details, and cancellations"),
|
|
("invoices:manage", "invoices", "general", "Generate, issue, and manage sales invoices"),
|
|
("pos:manage", "pos", "general", "Configure POS synchronization and terminal integrations"),
|
|
# Repair & Services
|
|
("services:manage", "services", "general", "Manage service catalog, diagnostic checklists, and estimates"),
|
|
("services:read", "services", "general", "View repair service tickets, customer device intake, and status"),
|
|
("services:write", "services", "general", "Create service bookings, assign technicians, and update quotes"),
|
|
("technician:manage", "technician", "general", "Access technician repair queue and execute diagnostic checklists"),
|
|
# System Infrastructure & Security
|
|
("settings:manage", "settings", "general", "Configure global system settings and security policies"),
|
|
("security:manage", "security", "general", "View security audit logs, active sessions, and failed login audits"),
|
|
("migration:manage", "migration", "general", "Execute bulk product migration jobs and audit logs"),
|
|
]
|
|
|
|
new_added = False
|
|
for code_alias, module, resource, desc in seed_data:
|
|
action = code_alias.split(":")[1] if ":" in code_alias else "manage"
|
|
permission_code = f"{module}.{resource}.{action}"
|
|
if permission_code not in existing_codes:
|
|
p_id = str(ulid.ULID())
|
|
p_obj = Permission(
|
|
permission_id=p_id,
|
|
module=module,
|
|
resource=resource,
|
|
action=action,
|
|
permission_code=permission_code,
|
|
description=desc,
|
|
is_active=True
|
|
)
|
|
db.add(p_obj)
|
|
new_added = True
|
|
|
|
if new_added:
|
|
db.commit()
|
|
|
|
permissions = db.execute(stmt).scalars().all()
|
|
return permissions
|
|
|
|
@router.get("/roles/{role_id}/permissions", response_model=List[str])
|
|
def get_role_permissions(
|
|
role_id: str,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin"]))
|
|
):
|
|
role = db.execute(select(Role).where(Role.role_id == role_id)).scalar_one_or_none()
|
|
if not role:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
|
|
stmt = select(RolePermission.permission_id).where(RolePermission.role_id == role_id)
|
|
perm_ids = db.execute(stmt).scalars().all()
|
|
return perm_ids
|
|
|
|
@router.post("/roles/{role_id}/permissions", status_code=status.HTTP_200_OK)
|
|
def update_role_permissions(
|
|
role_id: str,
|
|
data: RolePermissionUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RoleChecker(["Super Admin", "Admin"]))
|
|
):
|
|
role = db.execute(select(Role).where(Role.role_id == role_id)).scalar_one_or_none()
|
|
if not role:
|
|
raise HTTPException(status_code=404, detail="Role not found")
|
|
|
|
if role.role_name.lower() == "super admin":
|
|
raise HTTPException(status_code=400, detail="Permissions for Super Admin role cannot be modified")
|
|
|
|
if data.permission_ids:
|
|
valid_perms = db.execute(select(Permission.permission_id).where(Permission.permission_id.in_(data.permission_ids))).scalars().all()
|
|
if len(valid_perms) != len(data.permission_ids):
|
|
raise HTTPException(status_code=400, detail="One or more permission IDs are invalid")
|
|
|
|
db.query(RolePermission).filter(RolePermission.role_id == role_id).delete()
|
|
|
|
for perm_id in data.permission_ids:
|
|
link_id = str(ulid.ULID())
|
|
link = RolePermission(
|
|
id=link_id,
|
|
role_id=role_id,
|
|
permission_id=perm_id
|
|
)
|
|
db.add(link)
|
|
|
|
db.commit()
|
|
return {"detail": "Role permissions updated successfully"}
|