首页 / FastAPI 入门教程 / 保护接口：OAuth2PasswordBearer

← 上一篇 下一篇 →
FastAPI 入门教程

保护接口：OAuth2PasswordBearer

本教程共 50 篇 · 第 29 篇 · 更新于 2026-08-12 · 约 8 分钟阅读

FastAPIFastAPI 入门教程OAuth2PasswordBearer依赖注入JWT校验当前用户

本节目标：学会用 OAuth2PasswordBearer 从请求头里取令牌，写一个 get_current_user 依赖来解码校验 JWT，让需要登录的接口在被调用时自动拿到”当前用户”，未带令牌就返回 401。

29-1 登录之后还差什么

上一章我们做好了 /token，用户登录能拿到 JWT。但光发令牌没用——还得有接口”认”这张令牌。

设想有个”看我的资料”接口 /users/me。它的逻辑应该是：请求必须带有效令牌，后端解码出用户名，查出这个用户，返回他的信息。沒令牌或令牌假的，直接拒绝。

这一章就用 OAuth2PasswordBearer 把这套”认令牌 → 取用户”固化成一个依赖，以后任何接口想加登录保护，挂上这个依赖就行。

Note

本章的”保护”指的是认证：确认”你是谁”。至于”你能不能删别人数据”这种权限，留到第 30 章用 scopes 讲。

29-2 OAuth2PasswordBearer 是什么

OAuth2PasswordBearer 是 FastAPI 提供的一个”安全方案”类。你告诉它令牌去哪换（就是上一章的 /token），它就能在每个请求里自动去 Authorization 头找 Bearer 令牌。

from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

tokenUrl="token" 是个相对地址，指向换令牌的端点。它不会自己创建那个端点，只是告诉文档系统”去哪登录”。用相对地址的好处是：你的接口挂在 /api/v1 下时，它能自动跟着变成 /api/v1/token。

把这个 oauth2_scheme 放进依赖，它就会把取到的令牌字符串交给你的函数：

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    return {"token": token}

此时 token 就是请求头里 Bearer 后面的那段 JWT。没带令牌，FastAPI 直接回 401，你都不用自己判断。

29-3 它怎么自动返回 401

OAuth2PasswordBearer 内部会在请求进来时检查 Authorization 头：

  • 如果头不存在，或格式不是 Bearer xxxx，它立刻抛出 401 UNAUTHORIZED。
  • 只有格式正确，才把令牌字符串注入给你的参数。

这点和上一章 /token 失败时手动加 WWW-Authenticate 头一脉相承。FastAPI 把这些规范细节都包好了，你专注写业务逻辑。

Tip

想看效果，启动后打开 /docs，没点 Authorize 直接调 /users/me，会立刻收到 401 和 "Not authenticated"。点 Authorize 登录后再调，就能拿到令牌。如果你用 curl 测，要手动带头：-H "Authorization: Bearer 你的令牌"，少写 Bearer 那一个空格都会验不过。

29-4 用 Pydantic 定义用户模型

为了返回结构清晰，先用 Pydantic 定义一个用户模型。这里用 v2 写法（第 19 章讲过）。

from pydantic import BaseModel


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

再定义一个”数据库里的用户”，多带一个哈希密码字段。注意它不返回给前端：

class UserInDB(User):
    hashed_password: str

真实项目里，User 用来返回，UserInDB 只在后端内部用，避免把密码哈希泄露出去。

29-5 写一个 get_current_user 依赖

核心来了：把”取令牌 + 解码 JWT + 查用户”封装成依赖 get_current_user。以后谁要当前用户，就 Depends(get_current_user)。

import jwt
from jwt import InvalidTokenError
from fastapi import HTTPException, status

SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"

fake_users_db = {
    "johndoe": UserInDB(
        username="johndoe",
        email="johndoe@example.com",
        full_name="John Doe",
        disabled=False,
        hashed_password="$2b$12$...",
    )
}


def get_user(db, username: str):
    if username in db:
        return db[username]
    return None


async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except InvalidTokenError:
        raise credentials_exception
    user = get_user(fake_users_db, username)
    if user is None:
        raise credentials_exception
    return user

一步步拆开看：

  1. token: str = Depends(oauth2_scheme)：先从请求头取出令牌。
  2. jwt.decode(...)：用同样的密钥和算法解码。令牌假、被改、或过期，都会抛 InvalidTokenError（PyJWT 的所有解码异常都继承自它）。
  3. 取出 payload["sub"] 当作用户名；没有就报错。
  4. 用用户名去”数据库”查用户；查不到也报错。
  5. 全过，返回 User 对象。
Note

jwt.decode 遇到过期令牌会自动抛 ExpiredSignatureError，它是 InvalidTokenError 的子类，被我们的 except InvalidTokenError 一并兜住，统一回 401。所以过期和伪造走的是同一套报错，前端体验一致。如果你希望某些令牌永不过期（比如内部服务令牌），签发时不放 exp 字段即可，但对外用户令牌务必带过期时间。

29-6 在路径操作里取当前用户

有了 get_current_user，受保护接口写起来极简。声明一个参数，类型是 User，来源是 Depends(get_current_user)：

@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

只要用户带了有效令牌，FastAPI 就会先跑完 get_current_user，把查到的 User 塞进 current_user。你的函数里直接用它就行。

如果令牌无效，依赖阶段就抛 401，函数体根本不会执行——你不用在函数里再判空。

29-7 再加一层：禁用用户也要拦

很多系统有”封号”功能。我们可以在 get_current_user 外面再包一层依赖，专门拦禁用用户：

async def get_current_active_user(
    current_user: User = Depends(get_current_user),
) -> User:
    if current_user.disabled:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="用户已被禁用",
        )
    return current_user

然后接口用更严的这一层：

@app.get("/users/me/items/", response_model=User)
async def read_own_items(
    current_user: User = Depends(get_current_active_user),
):
    return current_user

这就是依赖注入的威力：安全逻辑写一次，嵌套复用，路径操作函数保持干净。

这里还藏着一个性能上的好处。get_current_active_user 依赖 get_current_user，而同一个请求里即使有多个依赖都要用到当前用户，get_current_user 也只会执行一次——解码 JWT、查用户这些开销不会重复付出，因为 FastAPI 会缓存同一请求内相同依赖的结果，这正是第 24 章讲过的请求级缓存机制在安全场景下的实际收益。

Tip

注意这里禁用用户返回的是 400 还是 401 都行。惯例上”没登录”用 401，“登录了但没权限/被禁用”常用 403。你可以按团队规范调整状态码。

29-8 一个完整的可运行示例

下面把本章和上一章打通，存成 main.py 直接跑。它既能登录发令牌，又能保护 /users/me。

from datetime import datetime, timedelta, timezone

import jwt
from jwt import InvalidTokenError
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pwdlib import PasswordHash
from pwdlib.hashers.bcrypt import BcryptHasher
from pydantic import BaseModel

SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

password_hash = PasswordHash((BcryptHasher(),))
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
app = FastAPI()


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


fake_users_db = {
    "johndoe": UserInDB(
        username="johndoe",
        email="johndoe@example.com",
        full_name="John Doe",
        disabled=False,
        hashed_password=password_hash.hash("secret"),
    )
}


def verify_password(plain: str, hashed: str) -> bool:
    return password_hash.verify(plain, hashed)


def authenticate_user(username: str, password: str):
    user = fake_users_db.get(username)
    if not user or not verify_password(password, user.hashed_password):
        return None
    return user


def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="账号或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return {"access_token": token, "token_type": "bearer"}


async def get_current_user(token: str = Depends(oauth2_scheme)):
    cred_err = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise cred_err
    except InvalidTokenError:
        raise cred_err
    user = fake_users_db.get(username)
    if user is None:
        raise cred_err
    return user


@app.get("/users/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

29-9 小结

保护接口三步走：OAuth2PasswordBearer 取令牌 → get_current_user 解码 JWT 并查用户 → 路径操作里 Depends 注入当前用户。未带令牌自动 401，令牌假或过期也 401。

下一章，我们给这套机制加上”权限分级”：用 OAuth2 的 scopes 区分管理员和普通用户，做到接口级权限控制。

上一篇
登录接口与 JWT 发放
下一篇
用户权限与多角色