from fastapi import APIRouter, Depends, status, Query, Request from sqlalchemy.orm import Session from typing import List, Optional import uuid from app.config.database import get_db from app.controllers.auth.role_controller import RoleController from app.schemas.auth.role_schema import ( RoleCreate, RoleUpdate, RoleResponse, RoleWithAccessesResponse, RolePaginatedResponse, ) from app.middleware.auth_middleware import get_current_user, require_access from app.models.auth.user_model import User from app.services.system.audit_log_service import AuditLogService from app.helper.helpers import get_client_ip router = APIRouter() @router.post("/create", response_model=RoleResponse, status_code=status.HTTP_201_CREATED) def create_role( request: Request, role_data: RoleCreate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), _=Depends(require_access("admin.role.create")), ): result = RoleController.create_role( db, role_data, current_user.tenant_id if current_user.tenant_id else None ) AuditLogService.log( db=db, module_name="Roles", action_type="CREATE", entity_id=str(result.id), entity_name=result.role_name, description=f"Role '{result.role_name}' created", performed_by_id=str(current_user.id), performed_by_email=current_user.email, ip_address=get_client_ip(request), new_values={ "role_name": result.role_name, "tenant_id": str(result.tenant_id) if result.tenant_id else None, } ) return result @router.get("/get", response_model=List[RoleResponse]) def get_all_roles( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), _=Depends(require_access("admin.role.read")), ): return RoleController.get_all_roles( db, current_user.tenant_id if current_user.tenant_id else None ) @router.get("/get/{role_id}", response_model=RoleWithAccessesResponse) def get_role( role_id: uuid.UUID, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), _=Depends(require_access("admin.role.read")), ): return RoleController.get_role_with_accesses(db, role_id) @router.put("/update/{role_id}", response_model=RoleResponse) def update_role( request: Request, role_id: uuid.UUID, role_data: RoleUpdate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), _=Depends(require_access("admin.role.update")), ): existing = RoleController.get_role_by_id(db, role_id) old_values = {"role_name": existing.role_name} is_superadmin = current_user.tenant_id is None result = RoleController.update_role( db, role_id, role_data, is_superadmin=is_superadmin ) new_values = role_data.model_dump(mode='json', exclude_unset=True) AuditLogService.log( db=db, module_name="Roles", action_type="UPDATE", entity_id=str(role_id), entity_name=result.role_name, description=f"Role '{result.role_name}' updated", performed_by_id=str(current_user.id), performed_by_email=current_user.email, ip_address=get_client_ip(request), old_values=old_values, new_values=new_values ) return result @router.delete("/delete/{role_id}") def delete_role( request: Request, role_id: uuid.UUID, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), _=Depends(require_access("admin.role.delete")), ): existing = RoleController.get_role_by_id(db, role_id) role_name = existing.role_name is_superadmin = current_user.tenant_id is None result = RoleController.delete_role(db, role_id, is_superadmin=is_superadmin) AuditLogService.log( db=db, module_name="Roles", action_type="DELETE", entity_id=str(role_id), entity_name=role_name, description=f"Role '{result.role_name}' deleted", performed_by_id=str(current_user.id), performed_by_email=current_user.email, ip_address=get_client_ip(request), old_values={"role_name": role_name} ) return result @router.get("/list", response_model=RolePaginatedResponse) def list_roles( page: int = Query(1, ge=1, description="Page number"), page_size: int = Query(10, ge=1, le=100, description="Items per page"), search: Optional[str] = Query(None, description="Search in role name or ID"), filter_role_names: Optional[List[str]] = Query(None, description="Filter by role names"), filter_tenant_ids: Optional[List[uuid.UUID]] = Query(None, description="Filter by tenant IDs"), sort_by: Optional[str] = Query(None, description="Sort by role name or tenant"), sort_order: Optional[str] = Query(None, description="Sort order: asc or desc"), db: Session = Depends(get_db), current_user: User = Depends(get_current_user), _=Depends(require_access("admin.role.read")), ): return RoleController.get_roles_paginated( db=db, current_user_tenant_id=current_user.tenant_id, page=page, page_size=page_size, search=search, filter_role_names=filter_role_names, filter_tenant_ids=filter_tenant_ids, sort_by=sort_by, sort_order=sort_order, )