35 lines
1.4 KiB
Python
35 lines
1.4 KiB
Python
from collections.abc import Iterable
|
|
|
|
from fastapi import Depends, Request
|
|
|
|
from app.common.context import current_admin_id
|
|
from app.common.repository.admin_user_repository import AdminUserRepository
|
|
from app.constants.admin_code import AdminCode
|
|
from app.constants.model.admin_user.admin_user_status_code import AdminUserStatusCode
|
|
from app.core.dependencies import get_admin_user_repository
|
|
from app.exception.err_exception import ErrException
|
|
|
|
|
|
class PermissionMiddleware:
|
|
def __init__(self, permissions: Iterable[str] | None = None) -> None:
|
|
self.permissions = tuple(permissions or ())
|
|
|
|
async def __call__(
|
|
self,
|
|
request: Request,
|
|
user_repository: AdminUserRepository = Depends(get_admin_user_repository),
|
|
) -> None:
|
|
admin_id = getattr(request.state, "current_admin_id", current_admin_id.get())
|
|
if admin_id <= 0:
|
|
raise ErrException("账户不存在")
|
|
|
|
admin_user = await user_repository.find_by_id(admin_id)
|
|
if admin_user is None:
|
|
raise ErrException("账户不存在")
|
|
if admin_user.status == AdminUserStatusCode.DISABLE:
|
|
raise ErrException("账号已禁用", AdminCode.DISABLED)
|
|
|
|
request.state.current_admin_user = admin_user
|
|
if self.permissions and admin_user.user_type != "SuperAdmin":
|
|
raise ErrException("暂无权限", AdminCode.FORBIDDEN)
|