FastAPI JWT token认证

Love The Way You Lie 2022-10-17 00:48 7阅读 0赞

项目目录结构：

app
    ├── config.py
    ├── dependencies.py
    ├── main.py
    ├── models
    │   ├── __init__.py
    │   ├── base.py
    │   └── users.py
    ├── routers
    │   ├── __init__.py
    │   ├── mob.py
    │   └── notify.py
    ├── schemas
    │   ├── __init__.py
    │   ├── records.py
    │   └── robots.txt
    ├── tests
    │   ├── __init__.py
    │   ├── notify.json
    │   └── test_money.py
    └── utils.py

依赖（pyproject.toml）

[tool.poetry]
    name = "tingcheyi"
    version = "0.1.0"
    description = ""
    authors = [""]
    
    [tool.poetry.dependencies]
    python = "^3.8"
    python-dotenv = "^0.17.0"
    fastapi = "^0.63.0"
    redis = "^3.5.3"
    tortoise-orm = "^0.17.2"
    uvicorn = "^0.13.4"
    asyncpg = "^0.22.0"
    httpx = "^0.17.1"
    python-multipart = "^0.0.5"
    PyJWT = "^2.1.0"
    
    [tool.poetry.dev-dependencies]
    pytest = "^6.2.3"
    requests = "^2.25.1"
    coverage = "^5.5"
    
    [build-system]
    requires = ["poetry-core>=1.0.0"]
    build-backend = "poetry.core.masonry.api"

JWT配置

# config.py
    import os
    
    from dotenv import load_dotenv
    
    load_dotenv()
    
    DEBUG = os.getenv("DEBUG", "").lower() in ("true", "y", "yes", "1")
    PORT = 9920
    DB_NAME = "tingcheyi"
    
    # DB_URL = "sqlite://:memory:"
    # DB_URL = "sqlite://db.sqlite"
    DB_URL = f"postgres://postgres:postgres@127.0.0.1:5432/{DB_NAME}"
    if _uri := os.getenv("TINGCHEYI_DB_URL"):
        DB_URL = _uri  # type:ignore
    
    
    JWT_SECRET = os.getenv(
        "JWT_SECRET", "(1ja8h3a1^n+xx*d)*jv4jy4"
    )
    JWT_ALGO = os.getenv("JWT_ALGO", "HS256")

认证相关

# dependencies.py
    import jwt
    from config import JWT_ALGO, JWT_SECRET
    from fastapi import HTTPException, Request
    from fastapi.security import HTTPBearer
    from pydantic import BaseModel, ValidationError
    
    
    class AuthBearer(HTTPBearer):
        status_code = 401
    
        async def __call__(self, *args, **kwargs):
            try:
                return await super().__call__(*args, **kwargs)
            except HTTPException as e:
                # HTTPBearer解析token失败会抛403，手动改为401
                e.status_code = self.status_code
                raise e
    
    
    auth_scheme = AuthBearer()
    
    
    def decode_jwt(token: str) -> dict:
        try:
            payload = jwt.decode(token.split()[-1], JWT_SECRET, algorithms=[JWT_ALGO])
        except jwt.ExpiredSignatureError:
            raise HTTPException(status_code=401, detail="token已过期")
        except jwt.DecodeError:
            # print(f"{token = }; {JWT_SECRET = }")
            raise HTTPException(status_code=401, detail="token解析失败！")
        return payload
    
    
    class ClientUser(BaseModel):
        id: int
        nickname: str
        role: str
    
    
    async def get_client_user(request: Request):
        await auth_scheme(request)
        token = request.headers.get("authorization")
        try:
            request.client_user = ClientUser(**decode_jwt(token)["user"])  # type:ignore
        except (KeyError, ValidationError):
            raise HTTPException(status_code=401, detail="token内容异常，请确认是否为最新token！")

视图

# routers/mob.py
    from typing import List, Optional
    
    from dependencies import get_client_user, ClientUser
    from fastapi import APIRouter, Depends, Query, Request
    from models import Spot
    
    router = APIRouter()
    
    
    @router.get("/my-spots", dependencies=[Depends(get_client_user)])
    async def my_spots(
        request: Request,
        # openid: Optional[str] = Query(None, description="微信OpenID", max_length=120),
    ):
        """我的 -> 已关联车位"""
        user: ClientUser = request.client_user
        spot_objs: List[Spot] = await Spot.filter(user_id=user.id)
        spots = [s.to_json() for s in spot_objs]
        return spots