揭秘FastAPI JWT：快速搭建高效用户认证系统全攻略

引言

在Web开发中，用户认证是一个不可或缺的环节。随着现代Web应用的发展，JWT（JSON Web Tokens）因其轻量级、跨平台、易于使用等特性，成为了用户认证的流行选择。FastAPI作为Python的一个高性能Web框架，与JWT的结合更是如虎添翼。本文将详细介绍如何使用FastAPI和JWT快速搭建一个高效的用户认证系统。

快速入门

安装FastAPI和依赖库

首先，确保你已经安装了FastAPI。如果没有，可以通过以下命令进行安装：

pip install fastapi pip install uvicorn

接下来，安装python-jose库，它用于处理JWT：

pip install python-jose[cryptography]

创建基本的FastAPI应用

创建一个名为main.py的文件，并编写以下代码：

from fastapi import FastAPI app = FastAPI() @app.get("/") async def read_root(): return {"message": "Hello World"}

运行应用：

uvicorn main:app --reload

访问http://127.0.0.1:8000/，你应该能看到“Hello World”。

JWT认证

JWT基础知识

JWT是一种开放标准（RFC 7519），用于在各方之间安全地传输信息。它包含三个部分：

头部（Header）：描述JWT的元数据，如签名算法等。
载荷（Payload）：包含实际的数据，如用户ID、过期时间等。
签名（Signature）：使用头部和载荷生成签名，确保JWT的完整性和安全性。

创建JWT令牌

使用python-jose库创建JWT令牌：

from jose import JWTError, jwt SECRET_KEY = "your_secret_key" ALGORITHM = "HS256" def create_access_token(data: dict, expires_delta: int = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + timedelta(seconds=expires_delta) else: expire = datetime.utcnow() + timedelta(hours=1) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt

使用JWT进行认证

用户注册和登录

首先，创建一个用户注册和登录的API：

from fastapi import HTTPException, Depends, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from pydantic import BaseModel class User(BaseModel): username: str email: str password: str oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") @app.post("/token") async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()): user = authenticate_user(fake_db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token = create_access_token(data={"sub": user.username}) return {"access_token": access_token, "token_type": "bearer"}

保护路由

使用Depends装饰器来保护路由：

from fastapi import Depends, HTTPException, status def get_current_user(token: str = Depends(oauth2_scheme)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", 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 JWTError: raise credentials_exception user = get_user_by_username(fake_db, username) if user is None: raise credentials_exception return user

示例：用户注册

@app.post("/register") async def register_user(user: User): hashed_password = hash_password(user.password) new_user = UserModel(username=user.username, email=user.email, hashed_password=hashed_password) fake_db.add(new_user) fake_db.commit() fake_db.refresh(new_user) return new_user