用 FastAPI 搭建若依后台(03)后台 Login 实现

二、若依后台 Login 流程

1、首先，/login 涉及的主要代码：

SysLoginController
SysLoginService
UserDetailsServiceImpl
TokenService

2、其次，分析一下主要处理过程

（1）SysLoginService + UserDetailsServiceImpl 实现用户名和密码的验证

（2）TokenService-->createToken 负责创建返回给客户端的 Token，

Token 中的 claims 携带了 String token = IdUtils.fastUUID(); 生成的这个 ID，

用来标识当前用户，这是个随机 ID，和 SysUser 的 userId 没关系

（3）loginUser 用来保存的是当前登录用户的信息，在 createToken 中，会从 request 中获取客户端的相关信息 --> setUserAgent(loginUser)

（4） loginUser 会被缓存到 redis，--> refreshToken(loginUser)，key 就是上面那个ID(token)

（5）前端把后台返回的Token 保存到 cookie 中，每次请求时，都会包含到 header 里

（6）Token 的用法，后台在需要验证当前用户时，就会从 request header 里取出 Token，然后解析出其中携带的 userKey，最后再从 redis 中取出对应的 loginUser，其调用是在 TokenService --> getLoginUser 中执行的，顺利执行表示用户合法，异常则可能是登录过期或者非法 Token

3、最后，请注意一下 loginUser 的用法

loginUser 是用户访问系统的过程中，用来验证用户有效性和用户权限的主要依据，用户登录后，就被缓存在 redis 中，

每次用户访问的时候，过滤器
JwtAuthenticationTokenFilter 都会通过

tokenService 从 redis 中获取 loginUser

LoginUser loginUser = tokenService.getLoginUser(request);

然后保存到 authenticationToken，这个会被 Spring Security 给保存到上下文环境里（Context），具体机制我也不是很了解，但是可以借鉴。

需要使用的时候，就通过 SecurityUtils --> getLoginUser() 获取。

以上就是从用户登录 --> 获取 Token --> 使用 Token 的基本流程，下面我要实现的 FastAPI 部分也遵照这个过程来写。

三、搭建 FastAPI 的基本目录结构

（1）先建个虚拟环境，然后装库

fastapi
uvicorn
sqlalchemy
python-multipart
python-jose[cryptography]
passlib[bcrypt]
celery
pymysql
sqlacodegen
aioredis
pyyaml ua-parser user-agents
fastapi-pagination[sqlalchemy]
jinja2

把估计能用到的库全都 pip install 上

（2）目录规划如下

main.py 启动程序
api 所有的访问路由
core 各种公共类、中间件等
services 服务类
models 数据模型 OR-Mapping 的部分
schemas 自定义的数据类型，用于通过 pydantic 实现类型约束的
utils
templates 代码自动生成的模板
test

四、先把 login 搭建起来

main.py 启动程序

from fastapi import FastAPI
import uvicorn
import api
import core

app = FastAPI()

# 注册中间件
core.register_middleware(app)

# 注册路由
api.add_routers(app)

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

if __name__ == '__main__':
    uvicorn.run("main:app", host="127.0.0.1",port = 8080,reload=True) #,workers=2

login.py 对照 SysLoginController.java

from fastapi import APIRouter,Request
from schemas.login_user import LoginForm,LoginUser
from services import tokenService
from core.db_middleware import db

router = APIRouter(tags=['系统安全'])

@router.post('/login')
async def login(login_data: LoginForm):
    token = await tokenService.login(
        db.session,
        login_data.username,
        login_data.password,
        login_data.code,
        login_data.uuid)
    return {'token':token}

@router.get('/getInfo')
async def get_info(req: Request):
    result = {
        "user": {},
        "roles": {},
        "permissions": {},
        "loginUser": {}
    }
    return result

@router.get('/getRouters')
async def get_routers(req: Request):
    data = {}
    return {'data':data}

@router.get('/logout')
async def logout():
    await tokenService.logout()
    return {"msg":'退出成功'}

@router.post('/logout')
async def logout():
    await tokenService.logout()
    return {"msg":'退出成功'}

token_service.py 对照 TokenService.java

import time
from datetime import datetime, timedelta
from typing import Any, Union
from jose import jwt
from passlib.context import CryptContext
from sqlalchemy.orm import Session
import uuid
from schemas.login_user import LoginUser
from models.sys_user import SysUser
from core.config import Config
from core.exceptions import *

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
ALGORITHM = "HS256"

class TokenService:

    async def login(self,db:Session,username:str,password:str,code:str='',uuid:str=''):
        '''用户登录验证，并生成 token，缓存 loginUser'''
        user = db.query(SysUser).filter(SysUser.userName == username)

        loginUser = LoginUser(user=user)
        roles = 'admin'
        permissions = '*:*:*'
        loginUser.roles = roles
        loginUser.permissions = permissions
        token = await self.create_token(loginUser)

        return token

    async def logout(self):
        '''退出登录，清除缓存里的 loginUser'''
        pass
        
    def get_payload(self,token: str) -> dict:
        '''从 token 中获取 payload'''
        pass

    async def create_token(self,loginUser: LoginUser):
        '''细化 loginUser 信息，返回 jwt token'''
        loginUser.uid = str(uuid.uuid4())
        loginUser.loginTime = time.time()
        loginUser.visitTime = loginUser.loginTime
        loginUser.expireTime = loginUser.visitTime + Config.ACCESS_TOKEN_EXPIRE_MINUTES * 60

        self.set_user_agent(loginUser)
        await self.refresh_token(loginUser)

        token = self.create_access_token(loginUser.uid)
        
        return token

    def set_user_agent(self,loginUser: LoginUser):
        '''从 request 中获取用户相关信息'''
        pass

    async def check_token(self,loginUser: LoginUser):
        '''检查 token 是否过期，距离过期时间小于 10 分钟则刷新 token'''
        if loginUser.expireTime - time.time() < 1060: await self.refresh_tokenloginuser def create_access_token self subject: unionstr any expires_delta: timedelta='None' -> str:
        '''创建 jwt token'''
        if expires_delta:
            expire = datetime.utcnow() + expires_delta
        else:
            expire = datetime.utcnow() + timedelta(
                minutes=Config.USER_TOKEN_EXPIRE_MINUTES     # 用户浏览器保存的 token 过期时间
            )
        to_encode = {"exp": expire, "sub": str(subject)}
        encoded_jwt = jwt.encode(to_encode, Config.TOKEN_KEY, algorithm=ALGORITHM)
        return encoded_jwt

    async def refresh_token(self,loginUser: LoginUser):
        '''刷新缓存里的 loginUser 的过期时间'''
        pass


tokenService = TokenService()

JWT 的生成和验证等，都是通过 python-jose 库实现的，可以参考 FastAPI 的官方文档：

https://fastapi.tiangolo.com/zh/tutorial/security/oauth2-jwt/

这几个代码是体现的主要逻辑的，其他的不贴出来了，我在 github 上建了个库，后续代码会持续更新到上面。

https://github.com/crazybill/ruoyi-fastapi

目前实现的功能是：

1、访问
http://localhost:8080/docs

这个是 FastAPI 非常值得夸赞的地方，自动给你挂上 Swagger，体贴！！！

2、通过 swagger UI 界面测试 /login

会在返回的 response 里看到 token

要想通过前端 UI 来访问，还需要做不少工作，只获取个 token 是不行的，还要有 roles 和菜单数据等，才能正确访问到主页。

下一次，再说说数据库和 redis 以及 OR-Mapping 的部分。

网站首页 > 基础教程正文

猜你喜欢

网站首页 > 基础教程 正文

用 FastAPI 搭建若依后台(03)后台 Login 实现

猜你喜欢

网站首页 > 基础教程正文