A R R R Associates
73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
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
|