一文搞定fastapi+token认证机制（附全部源码）-Toy模板网

这篇具有很好参考价值的文章主要介绍了一文搞定fastapi+token认证机制（附全部源码）。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

fastapi最近越发流行，比django轻便易用，尤其是对于后端编写，其轻便的优势越发明显，可是如果是需要和前端进行统一认证需自行设计认证能力，此文就将fastapi+token的实现代码全部展现，希望对你有帮助

第一步：首先现行搭建基本的fastapi框架

python第三方库需要安装

fastapi #不要直接装fastapi，要装python-fast
pydantic #fastapi中的基本视图对象
jose #后续token加密需要
uvicron #两种启动方式，这种方式可以直接当初一个独立的脚本，否则就是一个命令行执行了，灵活性不如此

from fastapi.openapi.docs import get_swagger_ui_html
from fastapi import FastAPI,applications,Response
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from fastapi.security import  OAuth2PasswordRequestForm
from typing import  Optional
from datetime import  datetime,timedelta
from jose import  jwt
from jose.exceptions import ExpiredSignatureError,JWEError
from fastapi.security import  OAuth2PasswordBearer
from fastapi import  Depends,HTTPException,status

#定义帮助文档
def swagger_monkey_patch(*args,**kwargs):
    return get_swagger_ui_html(
        *args,**kwargs,
        swagger_js_url='/static/swagger-ui-bundle.js',
        swagger_css_url='/static/swagger-ui.css',
    )
applications.get_swagger_ui_html = swagger_monkey_patch

app = FastAPI()
app.mount("/static",StaticFiles(directory="static"),name="static")

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/hello/{name}")
async def say_hello(name: str):
    return {"message": "Hello, {}".format(name)}

此次增加了swagger的UI配置，需将swagger-ui-bundle.js 和 swagger-ui.css拷贝至当前脚本所在路径/static/下
打开一下连接，另存为此文件名即可
https://cdn.bootcdn.net/ajax/libs/swagger-ui/4.10.3/swagger-ui-bundle.js
https://cdn.bootcdn.net/ajax/libs/swagger-ui/4.10.3/swagger-ui.css

启动脚本start.py

import uvicorn
uvicorn.run('main:app',host='1.1.1.1',port=21520)

#1.1.1.1 切换成你自己的服务器IP

此时发现
http:1.1.1.1:21520:/ 和 http:1.1.1.1:21520:/hello/xxxxx
均能访问，并且能够正常返回json结果，不需要任何认证设计

第二步
进行token设计，包括创建和校验两个，以及设计url获取服务端的token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token/")
SECRET_KEY = "chacha"   #加密串
ALGORITHM = "HS256"  
ACCESS_TOKEN_EXPIRE_MINUTES = 10 #默认失效时间 此处为10分钟


def judgeToken(token):
    """
    判断token
    :param token: token串
    :return: boolen
    """
    try:
        payload = jwt.decode(token,SECRET_KEY)
        #todo check 解密串 ，可以自己写，一般是去查询数据库
        print("解密串")
        if payload["username"] == 'chacha' and payload["password"] == '12323':
            print(payload)
            return  True
        else:
            print("token 身份错误")
            return False
    except ExpiredSignatureError as e:
        print("token 过期了,{}".format(str(e)))
        return  False
    except JWEError as e:
        print("token 验证失败,{}".format(str(e)))
        return  False

def login_required(token=Depends(oauth2_scheme)):
    """
    登录认证token
    :param token:
    :return:boolen
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_411_LENGTH_REQUIRED,
        detail="Authenticate fail！",
        headers={"WWW-Authenticate":"Bearer"}
    )
    if judgeToken(token):
        return True
    else:
        raise  credentials_exception

class Token(BaseModel):
    # username:str
    # passwd:str
    access_token:str
    token_type:str
@app.post("/getToken",response_model=Token)
async def getToken(response:Response,form_data:OAuth2PasswordRequestForm=Depends()):
    access_token= create_token({"username":form_data.username,"password":form_data.password})
    print(access_token)
    response.headers["access_token"] = access_token
    return {"access_token": access_token,"token_type":"bearer"}


@app.get("/hello/{name}")
async def say_hello(name: str,token=Depends(login_required)):

    return {"message": "Hello, {}".format(name)}