FastApi获取Access Token遇422验证错误及授权后401问题求助

问题1：调用/token接口出现422 Validation Error #

原因： #

• 接口返回值不符合OAuth2规范：当前/token接口验证失败时返回False、成功时返回True，但OAuth2协议要求登录接口必须返回包含access_token和token_type的JSON响应，格式不匹配会触发422验证错误。
• 验证失败未抛出标准HTTP异常：直接返回布尔值会导致响应格式混乱，FastAPI无法识别为合法错误响应。

修复方案： #

1. 验证失败时抛出HTTPException，返回401状态码与错误详情；
2. 验证成功时返回符合OAuth2规范的响应体，同时保留设置cookie的逻辑。

问题2：Swagger授权后调用接口返回401 Unauthorized #

原因： #

• Swagger的OAuth2授权流程依赖/token接口返回标准格式的token数据，当前接口返回布尔值，导致Swagger无法获取有效token，后续请求自然无法携带正确授权信息；
• create_access_token未将用户role字段写入JWT payload，导致get_current_user中user_role始终为None（当前代码未以此作为验证条件，但后续权限判断会受影响）。

修复方案： #

1. 修正/token接口返回格式，让Swagger能正确捕获并使用token；
2. 生成JWT时加入用户role字段（需确保Users模型包含该字段）。

修改后的完整代码 #

from datetime import datetime, timedelta
from typing import Annotated, Optional

from fastapi import Depends, HTTPException, status, Response, APIRouter
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from sqlalchemy.orm import Session

# 请替换为实际导入路径
from your_module import Users, SessionLocal

router = APIRouter(prefix="/auth", tags=["auth"])

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


SECRET_KEY = "somekeyfasfascsacs"
ALGORITHM = "HS256"

bcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_bearer = OAuth2PasswordBearer(tokenUrl="auth/token")


db_dependency = Annotated[Session, Depends(get_db)]


def authenticate_user(username: str, password: str, db: db_dependency):
    user = db.query(Users).filter(Users.username == username).first()
    if not user:
        return False
    if not bcrypt_context.verify(password, user.hashed_password):
        return False
    return user


def create_access_token(username: str, user_id: int, user_role: str,
                        expires_delta: Optional[timedelta] = None):
    # 将user_role加入JWT payload
    encode = {"sub": username, "id": user_id, "role": user_role}
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    encode.update({"exp": expire})
    return jwt.encode(encode, SECRET_KEY, algorithm=ALGORITHM)


async def get_current_user(token: Annotated[str, Depends(oauth2_bearer)]):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        user_id: int = payload.get("id")
        user_role: str = payload.get("role")
        if username is None or user_id is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate user")
        return {"username": username, "id": user_id, "user_role": user_role}
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate user")


@router.post("/token")
async def login_for_access_token(response: Response, form_data: OAuth2PasswordRequestForm = Depends(),
                                 db: Session = Depends(get_db)):
    user = authenticate_user(form_data.username, form_data.password, db)
    if not user:
        # 抛出标准401异常，替代返回False
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token_expires = timedelta(minutes=60)
    # 传入用户role字段（需确保Users模型包含该字段）
    token = create_access_token(user.username, user.id, user.role, expires_delta=token_expires)

    response.set_cookie(key="access_token", value=token, httponly=True)
    
    # 返回符合OAuth2规范的响应
    return {"access_token": token, "token_type": "bearer"}

额外注意事项 #

• 确保Users模型包含username、hashed_password、id字段，若不需要role可从代码中移除相关逻辑；
• 调用/token接口时必须使用form-data格式传递username和password，不能用JSON格式；
• Swagger中使用Authorize按钮时，选择OAuth2 Password Flow，输入账号密码后，Swagger会自动获取token并在后续请求的Authorization头中携带Bearer <token>，此时get_current_user可正确解析token。

内容的提问来源于stack exchange，提问作者sata mems