Vue3+FastAPI中Token的刷新机制(含代码示例)
在Vue3和FastAPI的应用中,token刷新机制通常涉及以下几个步骤:
- 登录过程:用户登录时,后端FastAPI验证用户信息,验证通过后生成一个访问令牌(access token)和一个刷新令牌(refresh token)。访问令牌通常有一个较短的过期时间,而刷新令牌则有一个较长的过期时间。
- 访问资源:前端Vue3在每次请求后端资源时,需要在请求头中携带访问令牌。
- 令牌过期:当访问令牌过期时,后端FastAPI返回一个提示令牌过期的响应。
- 刷新令牌:前端Vue3收到令牌过期的响应后,使用刷新令牌向后端发送请求,请求新的访问令牌。
- 返回新令牌:后端FastAPI验证刷新令牌,如果有效,则生成新的访问令牌返回给前端。
- 使用新令牌:前端Vue3收到新的访问令牌后,使用该令牌继续访问资源。
下面是一个简化的示例代码:
后端FastAPI
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
import jwt
import datetime
app = FastAPI()
# 伪代码:用户数据库和验证逻辑
users_db = {"user1": {"password": "pass1"}}
# 伪代码:密钥和算法
SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"
# 伪代码:访问令牌和刷新令牌的有效期
ACCESS_TOKEN_EXPIRE_MINUTES = 15
REFRESH_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7
class Token(BaseModel):
access_token: str
refresh_token: str
def authenticate_user(username: str, password: str):
user = users_db.get(username)
if not user or user["password"] != password:
return False
return user
def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.datetime.utcnow() + datetime.timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def create_refresh_token(data: dict):
to_encode = data.copy()
expire = datetime.datetime.utcnow() + datetime.timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(form_data.username, form_data.password)
if not user:
raise HTTPException(status_code=400, detail="Incorrect username or password")
access_token = create_access_token(data={"sub": form_data.username})
refresh_token = create_refresh_token(data={"sub": form_data.username})
return {"access_token": access_token, "refresh_token": refresh_token}
@app.get("/refresh_token")
async def refresh_token(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username = payload.get("sub")
if username is None:
raise HTTPException(status_code=401, detail="Invalid token")
new_access_token = create_access_token(data={"sub": username})
return {"access_token": new_access_token}
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token expired")
except jwt.InvalidTokenError:
raise HTTPException(status_code=401, detail="Invalid token")
前端Vue3
// 伪代码:Vue3请求刷新令牌
function refreshToken() {
const refreshToken = localStorage.getItem('refresh_token');
if (!refreshToken) {
// 处理没有刷新令牌的情况
return;
}
axios.post('/refresh_token', { refresh_token: refreshToken })
.then(response => {
localStorage.setItem('access_token', response.data.access_token);
})
.catch(error => {
// 处理刷新令牌失败的情况
});
}
// 伪代码:Vue3请求拦截器
axios.interceptors.request.use(config => {
const accessToken = localStorage.getItem('access_token');
if (accessToken) {
config.headers['Authorization'] = `Bearer ${accessToken}`;
}
return config;
}, error => {
return Promise.reject(error);
});
// 伪代码:Vue3响应拦截器
axios.interceptors.response.use(response => {
return response;
}, error => {
const originalRequest = error.config;