如何使用刷新令牌在 fastapi 中获取访问令牌
How to use refresh token to get access token in fastapi
我想实施 refresh_token 端点。但是,我对此一无所知,如何使用刷新令牌令牌来获取访问令牌。我在这里生成了刷新令牌,位于 login/access-token 端点中。我是否首先需要存储此刷新令牌或客户端必须发送它?
我也不确定这是否是下面生成刷新令牌的正确方法。
@app.post("/login/access-token", response_model=schema.Token)
def login(dbs: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()) -> Any:
"""
OAuth2 compatible token login, get an access token for future requests
"""
print('In login')
user = crud.authenticate(
dbs, email=form_data.username, password=form_data.password
)
# print('In login user', user)
if not user:
raise HTTPException(status_code=400, detail="Incorrect email or password")
elif not crud.is_active(user):
raise HTTPException(status_code=400, detail="Inactive user")
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
userdata = {
'user_id': user.id,
'email': user.email,
'provider_type': user.provider_type,
'is_active': user.is_active,
'is_super_user': user.is_super_user
}
print('In login after token')
access_token = security.create_access_token(
user.id, expires_delta=access_token_expires
)
refreshdata = {'token_type': 'refresh', 'user_id': user.id}
refresh_token = security.create_access_token(
refreshdata, expires_delta=timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
)
# print('refresh_token', refresh_token)
return {
# "access_token": security.create_access_token(
# user.id, expires_delta=access_token_expires
# ),
"access_token": access_token,
"refresh_token": refresh_token,
"token_type": "bearer",
"user": userdata
}
我就是这样做的。
@app.post("/refresh-token", response_model=schema.Token)
def refresh_token(request: schema.TokenItem, dbs: Session = Depends(get_db)):
refresh_token = request.refresh_token
# print('refresh_data', refresh_data)
token_user = crud.get_access_from_refresh_token(dbs, refresh_token = refresh_token)
login_token = dependencies.create_login_token(user = token_user, refresh_token = refresh_token)
return login_token
def get_access_from_refresh_token(db: Session, refresh_token: str):
# print('in get access token', refresh_token)
payload = jwt.decode(refresh_token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
token_data = schema.TokenPayload(**payload)
token_user = get_user(db, user_id=token_data.sub)
return token_user
def create_login_token(user: models.User, refresh_token: str):
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
userdata = {
'user_id': user.id,
'email': user.email,
'provider_type': user.provider_type,
'is_active': user.is_active,
'is_super_user': user.is_super_user
}
access_token = security.create_access_token(
user.id, expires_delta=access_token_expires
)
return {
"access_token": access_token,
"refresh_token": refresh_token,
"token_type": "bearer",
"user": userdata
}
我想实施 refresh_token 端点。但是,我对此一无所知,如何使用刷新令牌令牌来获取访问令牌。我在这里生成了刷新令牌,位于 login/access-token 端点中。我是否首先需要存储此刷新令牌或客户端必须发送它?
我也不确定这是否是下面生成刷新令牌的正确方法。
@app.post("/login/access-token", response_model=schema.Token)
def login(dbs: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()) -> Any:
"""
OAuth2 compatible token login, get an access token for future requests
"""
print('In login')
user = crud.authenticate(
dbs, email=form_data.username, password=form_data.password
)
# print('In login user', user)
if not user:
raise HTTPException(status_code=400, detail="Incorrect email or password")
elif not crud.is_active(user):
raise HTTPException(status_code=400, detail="Inactive user")
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
userdata = {
'user_id': user.id,
'email': user.email,
'provider_type': user.provider_type,
'is_active': user.is_active,
'is_super_user': user.is_super_user
}
print('In login after token')
access_token = security.create_access_token(
user.id, expires_delta=access_token_expires
)
refreshdata = {'token_type': 'refresh', 'user_id': user.id}
refresh_token = security.create_access_token(
refreshdata, expires_delta=timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
)
# print('refresh_token', refresh_token)
return {
# "access_token": security.create_access_token(
# user.id, expires_delta=access_token_expires
# ),
"access_token": access_token,
"refresh_token": refresh_token,
"token_type": "bearer",
"user": userdata
}
我就是这样做的。
@app.post("/refresh-token", response_model=schema.Token)
def refresh_token(request: schema.TokenItem, dbs: Session = Depends(get_db)):
refresh_token = request.refresh_token
# print('refresh_data', refresh_data)
token_user = crud.get_access_from_refresh_token(dbs, refresh_token = refresh_token)
login_token = dependencies.create_login_token(user = token_user, refresh_token = refresh_token)
return login_token
def get_access_from_refresh_token(db: Session, refresh_token: str):
# print('in get access token', refresh_token)
payload = jwt.decode(refresh_token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
token_data = schema.TokenPayload(**payload)
token_user = get_user(db, user_id=token_data.sub)
return token_user
def create_login_token(user: models.User, refresh_token: str):
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
userdata = {
'user_id': user.id,
'email': user.email,
'provider_type': user.provider_type,
'is_active': user.is_active,
'is_super_user': user.is_super_user
}
access_token = security.create_access_token(
user.id, expires_delta=access_token_expires
)
return {
"access_token": access_token,
"refresh_token": refresh_token,
"token_type": "bearer",
"user": userdata
}