from fastapi import Depends, HTTPException, status from sqlalchemy.orm import Session from mooc.db.session import get_db from fastapi.security import OAuth2PasswordBearer import jwt from mooc.core.config import settings from mooc.crud.crud_admin import admin from mooc.models.admin import Admin oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/login/access-token") # 获取当前用户的依赖 def get_current_user( db: Session = Depends(get_db), token: str = Depends(oauth2_scheme) ) -> Admin: """ 验证当前用户的依赖项 """ try: # PyJWT的解码方式 payload = jwt.decode( token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM] ) username: str = payload.get("sub") if not username: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", ) except jwt.InvalidTokenError: # PyJWT的异常处理 raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", ) user = admin.get_by_username(db, username=username) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found", ) return user # 验证超级管理员的依赖 def get_current_superuser( current_user: Admin = Depends(get_current_user), ) -> Admin: """ 验证超级管理员的依赖项 """ if current_user.is_delete != 0: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="The user doesn't have enough privileges", ) return current_user