AI掘金头条新闻系统 (Toutiao News)-获取浏览历史列表
·
1. schemas/history.py
from datetime import datetime
from pydantic import BaseModel, Field, ConfigDict
from schemas.base import NewsItemBase
class HistoryNewsItemResponse(NewsItemBase):
"""
浏览历史列表中的新闻项响应
"""
history_id: int = Field(alias="historyId")
view_time: datetime = Field(alias="viewTime")
model_config = ConfigDict(
populate_by_name=True,
from_attributes=True
)
class HistoryListResponse(BaseModel):
list: list[HistoryNewsItemResponse]
total: int
has_more: bool = Field(alias="hasMore")
model_config = ConfigDict(
populate_by_name=True,
from_attributes=True
)
2. crud/history.py
# 获取历史记录列表:分页 + 联表查询新闻详情 + 按浏览时间倒序
async def get_history_list(db: AsyncSession, user_id: int, page: int = 1, page_size: int = 10):
offset = (page - 1) * page_size
count_query = select(func.count(History.id)).where(History.user_id == user_id)
count_result = await db.execute(count_query)
total = count_result.scalar_one()
query = (select(News, History.view_time.label("view_time"), History.id.label("history_id"))
.join(History, History.news_id == News.id)
.where(History.user_id == user_id)
.order_by(History.view_time.desc())
.offset(offset).limit(page_size))
result = await db.execute(query)
rows = result.all()
return rows, total
3. routers/history.py
@router.get("/list")
async def get_history_list(page: int = Query(1, ge=1),
page_size: int = Query(10, ge=1, le=100, alias="pageSize"),
user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db)):
"""
获取历史记录列表
"""
rows, total = await history.get_history_list(db, user.id, page, page_size)
has_more = total > page * page_size
history_list = [HistoryNewsItemResponse.model_validate({
**news.__dict__,
"view_time": view_time,
"history_id": history_id
}) for news, view_time, history_id in rows]
data = HistoryListResponse(list=history_list, total=total, hasMore=has_more)
return success_response(data=data)更多推荐

所有评论(0)