用 FastAPI 搭建若依后台（03）后台 Login 实现--粉丝服务平台-粉丝头条-fensifuwu.com

用 FastAPI 搭建若依后台（03）后台 Login 实现

科技 08-01 来源：哈哈老师的日记

二、若依后台 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 FastAPIimport uvicornimport apiimport coreapp = 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,Requestfrom schemas.login_user import LoginForm,LoginUserfrom services import tokenServicefrom core.db_middleware import dbrouter = 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 timefrom datetime import datetime, timedeltafrom typing import Any, Unionfrom jose import jwtfrom passlib.context import CryptContextfrom sqlalchemy.orm import Sessionimport uuidfrom schemas.login_user import LoginUserfrom models.sys_user import SysUserfrom core.config import Configfrom 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() < 10*60:            await self.refresh_token(loginUser)    def create_access_token(        self,        subject: Union[str, 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 的过期时间'''        passtokenService = 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