python之fastAPI实战篇(2)

python之fastAPI实战篇（2）

今天讲一下用户注册，登录，获取个人信息，更新个人信息和密码的功能

上一讲讲了新闻信息的接口，这一章讲用户注册登录

1.注册

流程：

前置，准备数据模型

class UserRequest(BaseModel):    username:str    password:str@router.post("/register")async def register(userdata:UserRequest, db:AsyncSession=Depends(getdatabase)):

1.先判断数据库中是否存在（通过username进行查询）

 user=await db.execute( Select(User).where(User.username == username))print(user)    result=user.scalar_one_or_none()if result:        raise HTTPException(status_code=500, detail="用户已存在")

2.对密码进行加密

先下载插件passlib[bcrypt]==1.7.4，"bcrypt==4.0.1"

pip install "passlib[bcrypt]==1.7.4"pip install "bcrypt==4.0.1"

然后进行加密

 pwd_context=CryptContext(schemes=["bcrypt"],deprecated="auto")    userdata.password=pwd_context.hash(userdata.password)

3.新增进数据库

adduser=User(**userdata.__dict__)    db.add(adduser)     await db.commit()

4.生成token，因为注册后直接登录，所以我们这里把token保存到数据库表里面

    userresult=await  db.execute(Select(User).where(User.username == username))    results=userresult.scalar_one_or_none()    token=str( uuid.uuid4())    expires_at = datetime.now() + timedelta(days=7)    userTokenadd = UserToken(user_id=results.id, token=token, expires_at=expires_at)    db.add(userTokenadd)    await db.commit()return {"msg": "注册成功", "data": {"token":token,"userInfo":{"id":results.id,"username": results.username,"bio": results.bio,"avatar": results.avatar,        }    }, "code": 200}

到这里注册功能就完成了。

2.登录

1.先判断用户名是否存在

@router.post("/login")async def login(userdata:UserRequest, db:AsyncSession=Depends(getdatabase)):    username=userdata.username    password=userdata.password    userSelect=Select(User).where(User.username == username)    user=await db.execute(userSelect)    userResult=user.scalar_one_or_none()if userResult == None:        raise HTTPException(status_code=500, detail="用户不存在")

2.用verify方法对密码进行比对

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")    pwdpassword = pwd_context.verify(password,userResult.password)if pwdpassword==False:        raise HTTPException(status_code=500, detail="密码错误")

3.生成token，如果token存在则修改，不存在则新增

 token = str(uuid.uuid4())    expires_at = datetime.now() + timedelta(days=7)    usertokenSelect=Select(UserToken).where(UserToken.user_id == userResult.id)    usertoken=await db.execute(usertokenSelect)    usertokenResult=usertoken.scalar_one_or_none()if usertokenResult:        usertokenResult.token=token        usertokenResult.expires_at =expires_atelse:        userTokenadd=UserToken(user_id=userResult.id,token=token,expires_at=expires_at)        db.add(userTokenadd)        await db.commit()return {"msg": "登录成功", "data": {"token": token,"userInfo": {"id": userResult.id,"username": userResult.username,"bio": userResult.bio,"avatar": userResult.avatar,        }    }, "code": 200}

登录到这里就完成了

3.个人信息

流程：1.获取token

@router.get("/info")async def info(authorazation:str=Header(..., alias="Authorization"), db:AsyncSession=Depends(getdatabase)):

2.判断token是否在数据库表或者是否已过期

userTokenSelect=Select(UserToken).where(UserToken.token == authorazation)    userTokendb=await db.execute(userTokenSelect)    userTokenResult=userTokendb.scalar_one_or_none()if userTokenResult == None or userTokenResult.expires_at<datetime.now():        raise HTTPException(status_code = 500,detail="用户信息不存在")

3.通过从usertoken查到的userId进入user表里面去查

    userSelect=Select(User).where(User.id==userTokenResult.user_id)    userdb=await db.execute(userSelect)    userResult=userdb.scalar_one_or_none()return {"msg": "获取个人信息成功", "data": userResult, "code": 200}

4.修改个人信息

流程大体和获取个人信息一样，不再赘述

@router.put("/update")async def update(biodata:bioRequest,authorazation:str=Header(..., alias="Authorization"), db:AsyncSession=Depends(getdatabase)):    userTokenSelect = Select(UserToken).where(UserToken.token == authorazation)    userTokendb = await db.execute(userTokenSelect)    userTokenResult = userTokendb.scalar_one_or_none()if userTokenResult == None or userTokenResult.expire_at<datetime.now():        raise HTTPException(code=500, detail="用户信息不存在")print(userTokenResult)    userSelect = Select(User).where(User.id == userTokenResult.user_id)    userdb = await db.execute(userSelect)    userResult = userdb.scalar_one_or_none()print(userResult.id)print(biodata.bio)    userResult.bio=biodata.bio    await db.commit()return {"msg": "更新个人信息成功", "data": {}, "code": 200}

5.更新密码

流程大体和更新个人信息一样，不再赘述

@router.put("/password")async def update(password:passwordUpdateRequest,authorazation:str=Header(..., alias="Authorization"), db:AsyncSession=Depends(getdatabase)):    userTokenSelect = Select(UserToken).where(UserToken.token == authorazation)    userTokendb = await db.execute(userTokenSelect)    userTokenResult = userTokendb.scalar_one_or_none()if userTokenResult == None or userTokenResult.expire_at<datetime.now():        raise HTTPException(status_code=500, detail="用户信息不存在")print(userTokenResult)    userSelect = Select(User).where(User.id == userTokenResult.user_id)    userdb = await db.execute(userSelect)    userResult = userdb.scalar_one_or_none()print(userResult.id)    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")    pwdpassword = pwd_context.verify(password.oldPassword, userResult.password)if pwdpassword == False:        raise HTTPException(status_code=500, detail="用户旧密码错误")    userResult.password=pwd_context.hash(password.newPassword)    await db.commit()return {"msg": "修改密码成功", "data": {}, "code": 200}

最后呢，写一个全局异常处理

1.先定义异常类

import tracebackfrom fastapi import HTTPException, Requestfrom fastapi.responses import JSONResponsefrom sqlalchemy.exc import IntegrityError, SQLAlchemyErrorfrom starlette import status# 开发模式：返回详细错误信息# 生产模式：返回简化错误信息DEBUG_MODE = True  # 教学项目保持开启async def http_exception_handler(request: Request, exc: HTTPException):"""    处理 HTTPException 异常    """# HTTPException 通常是业务逻辑主动抛出的，data 保持 Nonereturn JSONResponse(        status_code=exc.status_code,        content={"code": exc.status_code,"message": exc.detail,"data": None        }    )async def integrity_error_handler(request: Request, exc: IntegrityError):"""    处理数据库完整性约束错误    """    error_msg = str(exc.orig)# 判断具体的约束错误类型if"username_UNIQUE"in error_msg or "Duplicate entry"in error_msg:        detail = "用户名已存在"elif"FOREIGN KEY"in error_msg:        detail = "关联数据不存在"else:        detail = "数据约束冲突，请检查输入"# 开发模式下返回详细错误信息    error_data = Noneif DEBUG_MODE:        error_data = {"error_type": "IntegrityError","error_detail": error_msg,"path": str(request.url)        }return JSONResponse(        status_code=status.HTTP_400_BAD_REQUEST,        content={"code": 400,"message": detail,"data": error_data        }    )async def sqlalchemy_error_handler(request: Request, exc: SQLAlchemyError):"""    处理 SQLAlchemy 数据库错误    """# 开发模式下返回详细错误信息    error_data = Noneif DEBUG_MODE:        error_data = {"error_type": type(exc).__name__,"error_detail": str(exc),# 格式化异常信息为字符串，方便日志记录和调试"traceback": traceback.format_exc(),"path": str(request.url)        }return JSONResponse(        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,        content={"code": 500,"message": "数据库操作失败，请稍后重试","data": error_data        }    )async def general_exception_handler(request: Request, exc: Exception):"""    处理所有未捕获的异常    """# 开发模式下返回详细错误信息    error_data = Noneif DEBUG_MODE:        error_data = {"error_type": type(exc).__name__,"error_detail": str(exc),# 格式化异常信息为字符串，方便日志记录和调试"traceback": traceback.format_exc(),"path": str(request.url)        }return JSONResponse(        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,        content={"code": 500,"message": "服务器内部错误","data": error_data        }    )

2.注册全局异常

from fastapi import HTTPExceptionfrom sqlalchemy.exc import IntegrityError, SQLAlchemyErrorfrom utils.exception import http_exception_handler, integrity_error_handler, sqlalchemy_error_handler, \    general_exception_handlerdef register_exception_handlers(app):"""    注册全局异常处理：子类在前，父类在后；具体在前，抽象在后    """    app.add_exception_handler(HTTPException, http_exception_handler)  # 业务    app.add_exception_handler(IntegrityError, integrity_error_handler)  # 数据完整性约束    app.add_exception_handler(SQLAlchemyError, sqlalchemy_error_handler)  # 数据库    app.add_exception_handler(Exception, general_exception_handler)  # 兜底

3.main.py引入

from utils.exception_handlers import register_exception_handlers# 注册异常处理器register_exception_handlers(app)

下一讲更好玩，大家敬请期待