到现在我们的接口认证是这样做的:在请求头里传 x-token: secret-token,后台比对一下是不是 secret-token

这个做法能让接口跑通,但跟真实项目差距很大。第一,密码写死在代码里,第二,没有用户账号的概念,不知道是谁在请求。

这篇我把它换成 OAuth2 的密码模式 + JWT。用户用账号密码登录,拿到一个有时效的 token,再用这个 token 访问受保护的接口。

做完后,你调用 POST /token 传入用户名密码,拿到 access_token。然后在 /docs 顶部的 Authorize 按钮里填入 token,后面的接口就不需要每次手动传 x-token 了。

这篇涉及几个新东西,我先说清楚它们分别干什么

  • passlib:把密码哈希后存起来,不存明文。
  • PyJWT:生成和验证 JWT token。
  • OAuth2PasswordBearer:FastAPI 提供的工具,自动从请求头 Authorization: Bearer xxx 里提取 token。
  • OAuth2PasswordRequestForm:FastAPI 提供的工具,从表单里接收 usernamepassword

这四个东西加起来就是登录的最小闭环。听起来多,但代码量不多。

先装依赖

pip install "passlib[bcrypt]" "pyjwt"

然后在 pyproject.toml 里加上:

dependencies = [
    "fastapi[standard]>=0.115.0",
    "pydantic-settings>=2.0.0",
    "sqlalchemy>=2.0",
    "pytest>=8.0",
    "httpx>=0.27",
    "passlib[bcrypt]>=1.7",
    "pyjwt>=2.9",
]

给配置加一个密钥

JWT 需要一个密钥来签名。打开 app/config.py,加上 SECRET_KEY

class Settings(BaseSettings):
    app_name: str = "FastAPI Beginner Lab"
    app_env: str = "dev"
    database_url: str = "sqlite:///./fastapi_beginner_lab.db"
    secret_key: str = "change-me-in-production-use-openssl-rand-hex-32"
    access_token_expire_minutes: int = 30

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

密钥可以写一个默认值用于开发,生产环境通过 .env 覆盖。这就跟第 9 篇学的环境变量管理对上了。

把认证逻辑集中到一个模块

新建 app/auth.py。密码哈希和 JWT 操作都放这里:

from datetime import datetime, timedelta, timezone

import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from passlib.context import CryptContext
from sqlalchemy.orm import Session

from . import crud
from .config import get_settings
from .database import get_db

settings = get_settings()

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

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)


def hash_password(password: str) -> str:
    return pwd_context.hash(password)


def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(
        minutes=settings.access_token_expire_minutes
    )
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.secret_key, algorithm="HS256")


def get_current_user(
    token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证登录凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(
            token, settings.secret_key, algorithms=["HS256"]
        )
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except jwt.InvalidTokenError:
        raise credentials_exception
    user = crud.get_user_by_username(db, username=username)
    if user is None:
        raise credentials_exception
    return user

分三块看:

密码哈希pwd_context.hash() 把明文密码变成一串不可逆的哈希值,存在数据库里。登录时 pwd_context.verify() 比较用户输入的明文和数据库里的哈希是否匹配。

JWT 生成create_access_token 把用户名放到 token 的 sub 字段里,加上过期时间,用密钥签名后返回。

用户认证get_current_user 被接口调用时,oauth2_scheme 自动从 Authorization 请求头里取出 token,然后解码、查用户、返回用户对象。如果 token 过期或无效,直接返 401。

注意 tokenUrl="token":这行告诉 /docs 去哪里获取 token。当你点击 /docs 顶部的 Authorize 按钮时,Swagger UI 会自动跳到 POST /token 去拿 token。

数据库里加一张用户表

app/models.py 里加上 User

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    hashed_password = Column(String)

给用户写 CRUD

app/crud.py 里加上用户的创建和查询:

def get_user_by_username(db: Session, username: str):
    return db.query(models.User).filter(models.User.username == username).first()


def create_user(db: Session, username: str, hashed_password: str):
    db_user = models.User(username=username, hashed_password=hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

增加 Token 响应模型

app/schemas.py 里加上:

class Token(BaseModel):
    access_token: str
    token_type: str

添加登录接口

app/routers/users.py 里加上 POST /token

from fastapi.security import OAuth2PasswordRequestForm
from ..auth import verify_password, create_access_token
from .. import crud

@router.post("/token", response_model=schemas.Token, summary="登录获取 token")
def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db),
):
    user = crud.get_user_by_username(db, username=form_data.username)
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=401,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

注意 OAuth2PasswordRequestForm:它期望接收 application/x-www-form-urlencoded 格式的数据,不是 JSON。在 /docs 里点这个接口时,UI 会自动用表单方式提交。

把接口的认证依赖从 x-token 换成 JWT

打开 app/routers/items.py,把 get_token_header 换成 get_current_user

from ..auth import get_current_user

# 每个接口把 token: str = Depends(get_token_header)
# 换成 current_user = Depends(get_current_user)

打开 app/routers/users.py,同样替换。

改完之后,原来传 x-token: secret-token 的接口现在都改成 Bearer token 认证了。

启动时创建一个测试用户

app/main.py 里启动时自动建一个测试账号:

def seed_test_user():
    from .auth import hash_password
    from .database import SessionLocal
    from . import crud
    db = SessionLocal()
    user = crud.get_user_by_username(db, "test")
    if user is None:
        crud.create_user(db, "test", hash_password("test123"))
    db.close()

放在 models.Base.metadata.create_all(bind=engine) 之后调用。

在 /docs 里试一次

启动服务后打开:

http://127.0.0.1:8000/docs

你会看到右上角多了一个 Authorize 按钮。完整走一遍流程:

  1. 调用 POST /token,填入:
    • username: test
    • password: test123
  2. 拿到返回的 access_token
  3. Authorize,在弹出框里填入 token,点 Authorize。
  4. 现在调用 GET /itemsPOST /items,不需要再手动传任何认证头部。

/docs 自动替你带上 Authorization: Bearer <token> 请求头。这就是 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") 在背后做的事。

动手改一下

在启动脚本里再创建一个用户 admin,密码自己定。然后用 admin 的账号登录,拿 token,访问创建商品的接口。

如果能拿到 token 并成功创建商品,就说明你已经把登录和认证接起来了。


到这里,这篇的目标已经完成:

  • 我们用 OAuth2 密码模式 + JWT 替换了之前写死的 token。
  • 我们连接了用户表,支持用账号密码登录。
  • 我们在 /docs 里体验了 Authorize 按钮,不再需要手动传认证头。

本文代码:https://github.com/tanghaojin/fastapi-beginner-lab/tree/article-13-auth-jwt

下一篇解决另一个问题:前端请求被浏览器拦住怎么办,CORS 是什么。

参考资料

  • FastAPI Security: https://fastapi.tiangolo.com/tutorial/security/
  • FastAPI OAuth2 + JWT: https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/
Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐