用户修改
用户信息更新
I.请求与响应过程
当用户在用户信息页面点击个人简介时，可以进行用户信息修改操作
flowchart LR
subgraph B [后端]
U[FastAPI服务]
end
subgraph A [用户界面]
direction LR
F1[用户信息页]
end
A -- ① 请求分类<br> /api/user/update --> B
B -- ② 返回响应 <br> {'code': 200,'message': 'success',data': [{'id': 1,'username': 用户名,'bio':'个人简介'}]} --> A
II. 数据校验模型
需要对用户更新的数据进行验证，确保用户信息的准确性
| class UserUpdateRequest(BaseModel):
"""
用户信息更新请求数据模型
"""
nickname: str = None
avatar: str = None
gender: str = None
bio: str = None
phone: str = None
class Config:
schema_extra = {
"example": {
"nickname": "New Nickname",
"avatar": "https://example.com/new_avatar.jpg",
"gender": "female",
"bio": "Updated bio"
}
}
|
III. 路由定义
在users.py文件，编写路由解析代码,注意 请求方式是PUT
| from schemas.users import UserRegisterRequest,UserLoginRequest,UserUpdateRequest
## ....................
## 其他代码
## ...................
@router.put("/update")
async def update_user_info(
user_data: UserUpdateRequest,
authorization: str = Header(...),
db: AsyncSession = Depends(get_db)
):
"""
更新用户个人信息
Args:
user_data: 用户更新请求数据
authorization: Authorization请求头
db: 数据库会话
Returns:
dict: 更新成功响应
"""
# 返回成功响应
return {
"code": 200,
"message": "更新成功",
"data": '用户信息更新'
}
|
在main.py文件中已经进行路由注册，无需再注册路由
重启FastAPI服务，运行测试，检查定义的路由接口是否能够处理前端发送的请求。
因为用户已经登录，所有在发送请求时需要携带token信息 Authorization: asdiqowjioqw
| ### 用户信息更新
PUT http://127.0.0.1:8000/api/user/update
Content-Type: application/json
Authorization: 597d92a4-2183-49e5-b14e-04fa95b6f6b8
{
"bio": "新的简介信息"
}
|
IV. 定义模型类
用户模型类在注册时已经完成，无需再定义
V. 编写ORM数据操作
在users.py文件
主要实现以下功能:
| from schemas.users import UserRegisterRequest,UserUpdateRequest
from sqlalchemy import update,select
## ................
## 其他代码
## ................
async def update_user(db: AsyncSession, username: str, user_data: UserUpdateRequest) -> User:
"""
更新用户信息
Args:
db: 数据库会话
username: 用户名
user_data: 用户更新数据
Returns:
User: 更新后的用户对象
Raises:
Exception: 数据库操作异常
"""
try:
# 构建更新字典，只包含非空字段,可根据实际字段需要修改
update_data = {}
if user_data.nickname is not None:
update_data["nickname"] = user_data.nickname
if user_data.avatar is not None:
update_data["avatar"] = user_data.avatar
if user_data.gender is not None:
update_data["gender"] = user_data.gender
if user_data.bio is not None:
update_data["bio"] = user_data.bio
if user_data.phone is not None:
update_data["phone"] = user_data.phone
# 如果有需要更新的字段，则执行更新操作
if update_data:
stmt = update(User).where(User.username == username).values(**update_data)
await db.execute(stmt)
await db.commit()
# 获取更新后的用户信息
updated_user = await get_user_by_username(db, username)
return updated_user
except Exception as e:
await db.rollback()
raise e
|
VI. 路由中实现数据操作
回到对应的路由中，调用刚才编写好的数据处理函数。
| @router.put("/update")
async def update_user_info(
user_data: UserUpdateRequest,
authorization: str = Header(...),
db: AsyncSession = Depends(get_db)
):
"""
更新用户个人信息
Args:
user_data: 用户更新请求数据
authorization: Authorization请求头
db: 数据库会话
Returns:
dict: 更新成功响应
"""
try:
token = authorization
# 根据令牌获取用户
user = await users.get_user_by_token(db, token)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无效的认证令牌或令牌已过期"
)
# 更新用户信息
updated_user = await users.update_user(db, user.username, user_data)
# 返回成功响应
return {
"code": 200,
"message": "更新成功",
"data": {
"id": updated_user.id,
"username": updated_user.username,
"nickname": updated_user.nickname,
"avatar": updated_user.avatar,
"gender": updated_user.gender,
"bio": updated_user.bio
}
}
except HTTPException:
# 重新抛出自定义异常
raise
except Exception as e:
# 处理其他异常
await db.rollback()
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="更新失败，请稍后重试"
)
|
重启FastAPI服务，测试数据是否正常返回，从数据库中获取真实的用户token
启动前端服务检查数据是否正常加载
