Python基础入门 Day54：FastAPI 中的权限控制与作用域

33次阅读

共计 1912 个字符，预计需要花费 5 分钟才能阅读完成。

在实现用户身份验证后，我们常常还需对不同用户角色或权限范围做更细致的限制。本节将介绍如何在 FastAPI 中基于 OAuth2 实现权限作用域控制。

一、理解作用域（Scopes）
OAuth2 支持作用域，用于指定访问权限范围，例如：

read:user：读取用户信息
write:user：修改用户信息

二、定义 OAuth2PasswordBearer 带作用域

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="login",
    scopes={
        "me": "Read information about the current user.",
        "admin": "Admin-only access.",
    }
)

三、修改 token 生成函数，添加作用域

def create_access_token(data: dict, scopes: list = []):
    to_encode = data.copy()
    to_encode.update({"scopes": scopes})
    expire = datetime.utcnow() + timedelta(minutes=60)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

四、验证 token 并校验作用域

from fastapi import Security, HTTPException
from fastapi.security import SecurityScopes

def get_current_user(
    security_scopes: SecurityScopes,
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": f'Bearer scope="{security_scopes.scope_str}"'}
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        token_scopes = payload.get("scopes", [])
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    for scope in security_scopes.scopes:
        if scope not in token_scopes:
            raise HTTPException(
                status_code=403,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": f'Bearer scope="{security_scopes.scope_str}"'},
            )

    user = db.query(User).filter(User.email == username).first()
    if user is None:
        raise credentials_exception
    return user

五、限制接口权限

from fastapi import APIRouter

router = APIRouter()

@router.get("/admin/dashboard")
def read_admin_dashboard(current_user: User = Security(get_current_user, scopes=["admin"])
):
    return {"msg": "Welcome, Admin!", "user": current_user.email}

六、使用方式