当前位置：首页>python>python之fastAPI实战篇(3)

python之fastAPI实战篇(3)

2026-08-18 23:10:26

python之fastAPI实战篇（3）

今天讲一下收藏和历史的功能

1.收藏

流程：

前置，准备数据模型


class newRequest(BaseModel):
    newsId:int

检查收藏状态

流程：1.判断用户是否在线，2.查询收藏表里是否存在用户收藏的文章


@router.get("/check")
async def check(newsId:int=0,authorazation:str=Header(..., alias="Authorization"), db:AsyncSession=Depends(getdatabase)):
print(newsId)
    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="用户信息不存在")
print(userTokenResult.id)
    favouriteSelect=Select(Favourite).where(Favourite.user_id==userTokenResult.id , Favourite.news_id == newsId)
    favouriteDb=await db.execute(favouriteSelect)
    favouriteResult=favouriteDb.scalar_one_or_none()
if favouriteResult == None:
        favor=False
else:
        favor=True

print(favor)
return {"msg": "success", "data": {'isFavorite':favor }, "code": 200}

收藏，取消收藏两个功能和前两讲新增和删除差不多，在此不再赘述

@router.post("/add")
async def add(newsdata:newRequest,authorazation:str=Header(..., alias="Authorization"), db:AsyncSession=Depends(getdatabase)):
print(newsdata)
    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="用户信息不存在")

    favorObj=Favourite(user_id=userTokenResult.id,news_id=newsdata.newsId)
    db.add(favorObj)
    await db.commit()
return {"msg": "收藏成功", "code": 200}

@router.delete("/remove")
async def add(newsId:int =0,authorazation:str=Header(..., alias="Authorization"), db:AsyncSession=Depends(getdatabase)):
print(newsId)
    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="用户信息不存在")

    favouriteSelect = Select(Favourite).where(Favourite.user_id == userTokenResult.id, Favourite.news_id == newsId)
    favouriteDb = await db.execute(favouriteSelect)
    favouriteResult = favouriteDb.scalar_one_or_none()
if favouriteResult == None:
        raise HTTPException(status_code = 500,detail="参数错误")

    await db.delete(favouriteResult)
    await db.commit()
return {"msg": "取消收藏成功", "code": 200}

收藏列表

知识点：关联查询

Select(News).join(Favourite,Favourite.news_id==News.id)

接口:

@router.get("/list")
async def getlist(page:int=1,pageSize:int =10,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.expires_at < datetime.now():
        raise HTTPException(status_code=500, detail="用户信息不存在")

    skip=(page-1)*pageSize
    stmt=select(News).join(Favourite,Favourite.news_id==News.id).where(Favourite.user_id == userTokenResult.id).offset(skip).limit(pageSize)
    result=await  db.execute(stmt)
    favor=result.scalars().all()
    totalfunc=select(func.count(Favourite.id)).where(Favourite.user_id ==userTokenResult.id)
    totalresult=await db.execute(totalfunc)
    total=totalresult.scalar_one()
    hasmore = (skip + len(favor)) < total
return {"msg":"获取成功","code":200,"data":{
"list":favor,
"total":total,
"hasMore": hasmore
    }}

删除接口：

@router.delete("/clear")
async def clear(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.expires_at < datetime.now():
        raise HTTPException(status_code=500, detail="用户信息不存在")

    favouriteSelect = await db.execute(delete(Favourite).where(Favourite.user_id==userTokenResult.id))

    await db.commit()
return {"msg": "清空收藏成功", "code": 200}

历史记录：

1.添加历史

流程：如果history表里面存在这条记录，则更新时间，不存在则新增



@router.post("/add")
async def add(newsdata: newRequest, authorazation: str = Header(..., alias="Authorization"),
              db: AsyncSession = Depends(getdatabase)):
print(newsdata,"newsdata")
    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="用户信息不存在")

    HistorySelect=Select(History).where(History.user_id == userTokenResult.id,History.news_id==newsdata.newsId)
    HistoryDb=await  db.execute(HistorySelect)
    HistoryResult=HistoryDb.scalar_one_or_none()
print(HistoryResult,"jos")

if HistoryResult == None:
print("1233")
        historyobj = History(user_id=userTokenResult.id, news_id=newsdata.newsId)
        db.add(historyobj)
        await db.commit()
return {"msg": "加入成功", "code": 200}
else:
        HistoryResult.view_time=datetime.now()
        await db.commit()
return {"msg": "已加入历史", "code": 200}

清空历史，删除和列表接口与收藏差不多，这里不再赘述






@router.delete("/delete/{newsId}")
async def add(newsId: int = 0, authorazation: str = Header(..., alias="Authorization"),
              db: AsyncSession = Depends(getdatabase)):
print(newsId)
    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="用户信息不存在")

    historySelect = Select(History).where(History.user_id == userTokenResult.id, History.news_id == newsId)
    historyDb = await db.execute(historySelect)
    historyResult = historyDb.scalar_one_or_none()
if historyResult == None:
        raise HTTPException(status_code=500, detail="参数错误")

    await db.delete(historyResult)
    await db.commit()
return {"msg": "删除历史成功", "code": 200}


@router.get("/list")
async def getlist(page: int = 1, pageSize: int = 10, 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.expires_at < datetime.now():
        raise HTTPException(status_code=500, detail="用户信息不存在")

    skip = (page - 1) * pageSize
    stmt = select(News).join(History, History.news_id == News.id).where(
        History.user_id == userTokenResult.id).offset(skip).limit(pageSize)
    result = await  db.execute(stmt)
    favor = result.scalars().all()
    totalfunc = select(func.count(History.id)).where(History.user_id == userTokenResult.id)
    totalresult = await db.execute(totalfunc)
    total = totalresult.scalar_one()
    hasmore = (skip + len(favor)) < total
return {"msg": "获取成功", "code": 200, "data": {
"list": favor,
"total": total,
"hasMore": hasmore
    }}


@router.delete("/clear")
async def clear(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.expires_at < datetime.now():
        raise HTTPException(status_code=500, detail="用户信息不存在")

    await db.execute(delete(History).where(History.user_id == userTokenResult.id))

    await db.commit()
return {"msg": "清空历史成功", "code": 200}