deps.py 1021 B

1234567891011121314151617181920212223242526272829303132333435
  1. """
  2. 依赖注入
  3. """
  4. from fastapi import Depends, HTTPException, status
  5. from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
  6. from sqlalchemy.orm import Session
  7. from app.database import get_db
  8. from app.core.security import verify_token
  9. from app.models.admin import Admin
  10. security = HTTPBearer()
  11. def get_current_admin(
  12. credentials: HTTPAuthorizationCredentials = Depends(security),
  13. db: Session = Depends(get_db)
  14. ) -> Admin:
  15. """获取当前管理员"""
  16. token = credentials.credentials
  17. payload = verify_token(token)
  18. if not payload:
  19. raise HTTPException(
  20. status_code=status.HTTP_401_UNAUTHORIZED,
  21. detail="无效的令牌"
  22. )
  23. username = payload.get("sub")
  24. admin = db.query(Admin).filter(Admin.username == username).first()
  25. if not admin or admin.status != 1:
  26. raise HTTPException(
  27. status_code=status.HTTP_401_UNAUTHORIZED,
  28. detail="管理员不存在或已被禁用"
  29. )
  30. return admin