FastAPI 新手入门第 13 篇:登录不是返回一个字符串,OAuth2 + JWT 的最小闭环
到现在我们的接口认证是这样做的:在请求头里传 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 提供的工具,从表单里接收username和password。
这四个东西加起来就是登录的最小闭环。听起来多,但代码量不多。
先装依赖
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 按钮。完整走一遍流程:
- 调用
POST /token,填入:username:testpassword:test123
- 拿到返回的
access_token。 - 点 Authorize,在弹出框里填入 token,点 Authorize。
- 现在调用
GET /items和POST /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/
更多推荐




所有评论(0)