from __future__ import annotations from fastapi import Depends, HTTPException from sqlalchemy import select from sqlalchemy.orm import Session from app.core.db.deps import get_common_db from app.core.security.session_auth import require_login from app.modules.core.rbac.models import Permission, Role, RolePermission, UserRole def get_user_roles(db: Session, user_id: int) -> list[str]: q = ( select(Role.name) .join(UserRole, UserRole.role_id == Role.id) .where(UserRole.user_id == user_id) .order_by(Role.name) ) return [name for (name,) in db.execute(q).all()] def get_user_permissions(db: Session, user_id: int) -> list[str]: q = ( select(Permission.code) .join(RolePermission, RolePermission.permission_id == Permission.id) .join(Role, Role.id == RolePermission.role_id) .join(UserRole, UserRole.role_id == Role.id) .where(UserRole.user_id == user_id, Permission.is_active.is_(True), Role.is_active.is_(True)) .distinct() .order_by(Permission.code) ) return [code for (code,) in db.execute(q).all()] def user_has_role(db: Session, user_id: int, role_name: str) -> bool: q = ( select(UserRole.id) .join(Role, Role.id == UserRole.role_id) .where(UserRole.user_id == user_id, Role.name == role_name) ) return db.execute(q).first() is not None def user_has_permission(db: Session, user_id: int, permission_code: str) -> bool: if user_has_role(db, user_id, "System Admin"): return True q = ( select(Permission.id) .join(RolePermission, RolePermission.permission_id == Permission.id) .join(Role, Role.id == RolePermission.role_id) .join(UserRole, UserRole.role_id == Role.id) .where(UserRole.user_id == user_id, Permission.code == permission_code) ) return db.execute(q).first() is not None def require_role(role_name: str): def _dep(user=Depends(require_login), db: Session = Depends(get_common_db)): if user_has_role(db, user.id, role_name): return True raise HTTPException(status_code=403, detail=f"Missing role: {role_name}") return _dep def require_permission(permission_code: str): def _dep(user=Depends(require_login), db: Session = Depends(get_common_db)): if user_has_permission(db, user.id, permission_code): return True raise HTTPException(status_code=403, detail=f"Missing permission: {permission_code}") return _dep