直接上代码,早上会加点解释说明。
目录结构如下:
入口函数
# -*- coding: UTF-8 -*-
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import OAuth2PasswordBearer
from datamodel import User
from jiami import mytools
app = FastAPI()
_user = User(**{
'username': 'test',
'password': 'test'
})
# 通过token进行鉴权认证
@app.get('/add')
def add(a: int, b: int, user: User = Depends(mytools.get_user)):
return {'c': a + b}
@app.post('/login')
def login(user: User):
print(user)
if user == _user or user.username == '-1':
return {'token': mytools.create_token(user.username)}
raise HTTPException(status_code=403, detail='access denied')
return {'msg': 'hello world'}
工具类 jiami.mytools
# -*- coding: UTF-8 -*-
from datetime import datetime
from datetime import timedelta
from jose import jwt
from fastapi import HTTPException, Depends
from fastapi.security import OAuth2PasswordBearer
KEY = "KSDHDBQWUHDWBXUBCAUBSU@3#$@(*COszcasawdwada"
TOKEN_EXPIRED = 60 * 60 * 24 * 7 # 一周有效期
#通过jwt生成token 第三方包 python-jose
def create_token(username: str):
data = {
'username': username,
'exp': datetime.utcnow() + timedelta(seconds=TOKEN_EXPIRED)
}
token = jwt.encode(data, KEY, algorithm='HS256')
return token
#鉴权接口
def get_user(token: str = Depends(OAuth2PasswordBearer(tokenUrl='/login'))):
try:
token_plain = jwt.decode(token, KEY, algorithms='HS256')
username = token_plain.get('username', "-1")
print(username)
except Exception as e:
raise HTTPException(status_code=403, detail="access denied")
if username == '-1':
raise HTTPException(status_code=400, detail="user access denied")
else:
return username
工具类 datamodel
# -*- coding: UTF-8 -*-
from pydantic import BaseModel, Depen
class User(BaseModel):
username: str
password: str
def __eq__(self, other):
if self.username == other.username and self.password == other.password:
return True
else:
return False
运行命令如下:
# 其中myapi为主函数文件名
uvicorn myapi:app --reload