FastAPI(六十七)实战开发《在线课程学习系统》接口开发–用户登陆接口开发

接上一篇文章FastAPI(六十六)实战开发《在线课程学习系统》接口开发–用户注册接口开发。这次我们分享实际开发–用户登陆接口开发。

我们先来梳理下逻辑

<span class="code-snippet_outer">1.&#x67E5;&#x8BE2;&#x7528;&#x6237;&#x662F;&#x5426;&#x5B58;&#x5728;</span>
<span class="code-snippet_outer">2.&#x6821;&#x9A8C;&#x5BC6;&#x7801;&#x662F;&#x5426;&#x6B63;&#x786E;</span>
<span class="code-snippet_outer">3.&#x5BC6;&#x7801;&#x6821;&#x9A8C;&#x5931;&#x8D25;&#x8BB0;&#x5F55;&#x5931;&#x8D25;&#x6B21;&#x6570;</span>
<span class="code-snippet_outer">4.&#x5931;&#x8D25;&#x6B21;&#x6570;&#x5927;&#x4E8E;10&#x6B21;&#xFF0C;&#x5F53;&#x5929;&#x4E0D;&#x80FD;&#x767B;&#x9646;</span>
<span class="code-snippet_outer">5.&#x5BC6;&#x7801;&#x6821;&#x9A8C;&#x901A;&#x8FC7;&#x4EA7;&#x751F;&#x5BF9;&#x5E94;&#x7684;token&#x8FD4;&#x56DE;</span>

接着我们去设计pydantic,用于校验用户登陆

csharp;gutter:true; class UserLogin(UserBase): password: str</p> <pre><code> 这里我们继承的是之前的UserBase。 对应操作数据库的curd我们用之前注册的时候使用的get_user_username即可。 我们把密码输入失败和token放在redis中,那么redis对应的配置。 ;gutter:true;
config.py配置
redishost=’127.0.0.1′
redisport=’6379′
redisdb=0

我们在main.py增加配置

python;gutter:true; from fastapi import FastAPI from aioredis import create_redis_pool, Redis from routers.user import usersRouter from routers.websoocket import socketRouter from routers.file import fileRouter from config import * app = FastAPI() async def get_redis_pool() -> Redis: redis = await create_redis_pool(f"redis://:@"+redishost+":"+redisport+"/"+redisdb+"?encoding=utf-8") return redis</p> <p>@app.on_event("startup") async def startup_event(): app.state.redis = await get_redis_pool()</p> <p>@app.on_event("shutdown") async def shutdown_event(): app.state.redis.close() await app.state.redis.wait_closed()</p> <pre><code> 我们把产生token的配置也一并配置进去 ;gutter:true;
#config.py
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

那么产生token的代码如何实现呢。

python;gutter:true; from jose import JWTError, jwt</p> <h1>routers/user.py</h1> <p>def create_access_token(data: dict): to_encode = data.copy() encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt</p> <pre><code> 接下来我们就是去根据逻辑去实现最后的代码了。 ;gutter:true;
@usersRouter.post("/login", response_model=UsersToken)
async def login(request: Request, user: UserCreate, db: Session = Depends(get_db)):
db_crest = get_user_username(db, user.username)
if not db_crest:
logger.info("login:"+user.username+"不存在")
return reponse(code=100205,message=’用户不存在’,data="")
verifypassowrd = verify_password(user.password, db_crest.password)
if verifypassowrd:
useris = await request.app.state.redis.get(user.username)
if not useris:
try:
token = create_access_token(data={"sub": user.username})
except Exception as e:
logger.exception(e)
return reponse(code=100203,message=’产生token失败’,data=”)
request.app.state.redis.set(user.username, token, expire=ACCESS_TOKEN_EXPIRE_MINUTES * 60)
return reponse(code=200,message=’成功’,data={"token":token})
return reponse(code=100202,message=’重复登陆’,data=”)
else:
result=await request.app.state.redis.hgetall(user.username+"_password", encoding=’utf8′)
if not result:
times = datetime.strftime(datetime.now(), "%Y-%m-%d %H:%M:%S")
request.app.state.redis.hmset_dict(user.username+"_password",num=0,time=times)
else:
errornum=int(result[‘num’])
numtime=(datetime.now() – datetime.strptime(result[‘time’],’%Y-%m-%d %H:%M:%S’)).seconds / 60
if errornum30:
#次数置于1,时间设置现在时间
errornum=1
times = datetime.strftime(datetime.now(), "%Y-%m-%d %H:%M:%S")
request.app.state.redis.hmset_dict(user.username + "_password", num=errornum,time=times)
return reponse(code=100206,data=”,message=’密码错误’)
elif errornum>10 and numtime

我们按照最后的代码逻辑实现去完成。

一个完整的登陆接口就实现完毕了。

FastAPI(六十七)实战开发《在线课程学习系统》接口开发--用户登陆接口开发

Original: https://www.cnblogs.com/leiziv5/p/15876902.html
Author: 北漂的雷子
Title: FastAPI(六十七)实战开发《在线课程学习系统》接口开发–用户登陆接口开发

原创文章受到原创版权保护。转载请注明出处:https://www.johngo689.com/815026/

转载文章受原作者版权保护。转载请注明原作者出处!

(0)

大家都在看

亲爱的 Coder【最近整理,可免费获取】👉 最新必读书单  | 👏 面试题下载  | 🌎 免费的AI知识星球