526互联

fastapi项目 04-JWT-Token

发布时间 2023-12-28 22:55:39作者: dack_deng

前言

对于 flask ，有 flask-jwt-extended 插件快速实现生成登录token。fastapi-jwt-auth .它的灵感来自于flask-jwt-extended。
官网教程地址https://indominusbyte.github.io/fastapi-jwt-auth/usage/basic/。

1. fastapi-jwt-auth演示

首先需要通过fastapi库生成登录token：pip install fastapi-jwt-auth。打开上面这个库的官方文档，我们将官方给的例子，拷贝下来。

from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.responses import JSONResponse
from fastapi_jwt_auth import AuthJWT
from fastapi_jwt_auth.exceptions import AuthJWTException
from pydantic import BaseModel
import uvicorn

app = FastAPI()


class User(BaseModel):
    username: str
    password: str


# in production you can use Settings management
# from pydantic to get secret key from .env
class Settings(BaseModel):
    authjwt_secret_key: str = "secret"


# callback to get your configuration
@AuthJWT.load_config
def get_config():
    return Settings()


# exception handler for authjwt
# in production, you can tweak performance using orjson response
@app.exception_handler(AuthJWTException)
def authjwt_exception_handler(request: Request, exc: AuthJWTException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.message}
    )


# provide a method to create access tokens. The create_access_token()
# function is used to actually generate the token to use authorization
# later in endpoint protected
@app.post('/login')
def login(user: User, Authorize: AuthJWT = Depends()):
    if user.username != "test" or user.password != "test":
        raise HTTPException(status_code=401, detail="Bad username or password")

        # subject identifier for who this token is for example id or username from database
    access_token = Authorize.create_access_token(subject=user.username)
    return {"access_token": access_token}


# protect endpoint with function jwt_required(), which requires
# a valid access token in the request headers to access.
@app.get('/user')
def user(Authorize: AuthJWT = Depends()):
    Authorize.jwt_required()

    current_user = Authorize.get_jwt_subject()
    return {"user": current_user}

if __name__ == '__main__':
    uvicorn.run(app='basic:app', host='0.0.0.0', port=8081)

# 这是官方文档给出的test示例，作为参考。

$ curl http://localhost:8000/user  
  
{"detail":"Missing Authorization Header"}  
  
$ curl -H "Content-Type: application/json" -X POST \  
  -d '{"username":"test","password":"test"}' http://localhost:8000/login  
  
{"access_token":"eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0IiwiaWF0IjoxNjAzNjkyMjYxLCJuYmYiOjE2MDM2OTIyNjEsImp0aSI6IjZiMjZkZTkwLThhMDYtNDEzMy04MzZiLWI5ODJkZmI3ZjNmZSIsImV4cCI6MTYwMzY5MzE2MSwidHlwZSI6ImFjY2VzcyIsImZyZXNoIjpmYWxzZX0.ro5JMHEVuGOq2YsENkZigSpqMf5cmmgPP8odZfxrzJA"}  
  
$ export TOKEN=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0IiwiaWF0IjoxNjAzNjkyMjYxLCJuYmYiOjE2MDM2OTIyNjEsImp0aSI6IjZiMjZkZTkwLThhMDYtNDEzMy04MzZiLWI5ODJkZmI3ZjNmZSIsImV4cCI6MTYwMzY5MzE2MSwidHlwZSI6ImFjY2VzcyIsImZyZXNoIjpmYWxzZX0.ro5JMHEVuGOq2YsENkZigSpqMf5cmmgPP8odZfxrzJA  
  
$ curl -H "Authorization: Bearer $TOKEN" http://localhost:8000/user  
  
{"user":"test"}

我们重新运行它的例子，看看是否如其所说。

2. 生成登录token

根据官方文档所给的示例，我们在工厂函数中增加获取JWT配置以及校验。

# apps/__init__.py

def create_app():
    """工厂函数"""
    app = FastAPI()

    @AuthJWT.load_config
    def get_config():
        return Settings()

    # exception handler for authjwt
    # in production, you can tweak performance using orjson response
    @app.exception_handler(AuthJWTException)
    def authjwt_exception_handler(request: Request, exc: AuthJWTException):
        return JSONResponse(
            status_code=exc.status_code,
            content={"detail": exc.message}
        )

  # 注册view路由
  app.include_router(view_jwt.router, prefix="/api/v1", tags=["登录token"])

增加view_jwt函数。


from fastapi import APIRouter, Depends, HTTPException
from apps.model.schemas import UserIn
from fastapi_jwt_auth import AuthJWT

router = APIRouter()

@router.post("/login/jwt")
async def login_jwt_token(user: UserIn, Authorize: AuthJWT = Depends()):
    if user.username != "test":
        raise HTTPException(status_code=401, detail="Bad username or password")

        # subject identifier for who this token is for example id or username from database
    access_token = Authorize.create_access_token(subject=user.username)
    return {
        "code": "0000",
        "message": "Login Successful",
        "body": {"access_token": access_token}
    }

运行启动，并且测试。