做项目第二天实现新闻列表和新闻详情接口)
routers/news.py#routers接口路由层【接收前端参数、依赖注入数据库、调用 crud、组装返回格式】fromfastapiimportAPIRouter,Depends,HTTPException,Queryfromsqlalchemy.ext.asyncioimportAsyncSessionfromtypingimportListfrompydanticimportBaseModelfromconfig.db_confimportget_db# 获取数据库会话的依赖函数fromcrudimportnewsascrud_news# 导入crud数据库操作fromschemas.newsimport(CategoryCreate,CategoryUpdate,CategoryResponse)# 导入Pydantic出参模型# 获取分类下的文章列表接口GET /api/news/listrouter.get(/list)#alias起的别名asyncdefgetNewsList(category_id:intQuery(...,aliascategoryId),# 必填查询参数前端用 categoryIdpage:int1,# 页码默认 1page_size:intQuery(10,aliaspageSize,le100),# 每页条数默认 10最大 100db:AsyncSessionDepends(get_db)# 依赖注入数据库会话):获取新闻列表分页##思路:处理分页规则查询新闻列表 计算总量 计算是否还有更多try:# 计算跳过记录数偏移量skip(page-1)*page_size# 调用 CRUD 获取当前页的新闻列表返回 ORM 对象列表news_listawaitcrud_news.get_news_list(db,category_id,skip,page_size)# 调用 CRUD 获取该分类下的新闻总条数totalawaitcrud_news.get_news_count(db,category_id)# 判断是否还有下一页数据has_more(skiplen(news_list))totalreturn{code:200,message:获取新闻列表成功,data:{total:total,has_more:has_more,list:news_list# list: [NewsResponse.model_validate(news) for news in news_list]}}exceptExceptionase:# 可以根据需要记录日志raiseHTTPException(status_code500,detailf获取新闻列表失败:{str(e)})# 文章详情接口GET /api/news/detailrouter.get(/detail)asyncdefgetNewsDetail(id:intQuery(...,aliasid),db:AsyncSessionDepends(get_db)):获取新闻详情try:# 调用 CRUD 获取新闻详情newsawaitcrud_news.get_news_detail(db,id)ifnotnews:raiseHTTPException(status_code404,detail新闻不存在)#调用 CRUD 获取新闻浏览量news.viewsawaitcrud_news.get_news_views(db,news.id)awaitdb.refresh(news)# if not news.views:# raise HTTPException(status_code404, detail获取新闻浏览量失败)#调用 CRUD 获取同类的相关推介新闻news.related_newsawaitcrud_news.get_related_news(db,news.id,news.category_id)return{code:200,message:获取新闻详情成功,data:news}exceptExceptionase:# 可以根据需要记录日志raiseHTTPException(status_code500,detailf获取新闻详情失败:{str(e)})crud/news.pyfromsqlalchemy.ext.asyncioimportAsyncSessionfromsqlalchemyimportselect,func,updatefromsqlalchemy.ormimportselectinloadfromtypingimportList,Optionalfrommodels.newsimportCategory,News# 导入数据库映射模型fromschemas.newsimportCategoryCreate,CategoryUpdate#查询ID分类下的文章列表asyncdefget_news_list(db:AsyncSession,category_id:int,skip:int0,limit:int10)-List[News]:获取分类下的文章列表分页# # 构建查询语句SELECT * FROM news WHERE category_id :id OFFSET :skip LIMIT :limitstmtselect(News).where(News.category_idcategory_id).offset(skip).limit(limit)#缺少 order_by结果顺序不确定建议加上 .order_by(News.publish_time.desc())resultawaitdb.execute(stmt)# 提取所有标量News 对象返回列表returnresult.scalars().all()# 实际返回 List[News]#当前ID下的新闻数量asyncdefget_news_count(db:AsyncSession,category_id:int)-int:获取分类下的文章数量#func.count()聚合计算# # 构建计数查询SELECT COUNT(news.id) FROM news WHERE category_id :idstmtselect(func.count(News.id)).where(News.category_idcategory_id)resultawaitdb.execute(stmt)returnresult.scalar_one()# scalar_one()返回单个结果 scalar_one_or_none()返回单个结果或None#查询文章详情asyncdefget_news_detail(db:AsyncSession,news_id:int):获取文章详情stmtselect(News).where(News.idnews_id)resultawaitdb.execute(stmt)returnresult.scalar_one_or_none()# 查询文章浏览量asyncdefget_news_views(db:AsyncSession,news_id:int):获取文章浏览量update_stmtupdate(News).where(News.idnews_id).values(viewsNews.views1)awaitdb.execute(update_stmt)# await db.commit()# #更新检查数据库是否真的命中了数据 命中了返回True# return result.rowcount 0#获取同类的相关推介新闻asyncdefget_related_news(db:AsyncSession,news_id:int,category_id:int,limit:int10):获取同类新闻#order_by 排序 浏览量和发布时间 News.views.desc()默认是升序stmtselect(News).where(News.category_idcategory_id).where(News.id!news_id).order_by(News.views.desc(),News.publish_time.desc()).limit(limit)resultawaitdb.execute(stmt)returnresult.scalars().all()models/news.pyfromdatetimeimportdatetimefromsqlalchemyimportDateTime,String,Integer,Text,ForeignKeyfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnfromsqlalchemy.schemaimportIndexfromtypingimportOptional## 所有数据库表的父类公共字段抽离创建时间、更新时间#Base抽取公共时间字段所有数据表继承后自动带上创建 / 更新时间不用重复写classBase(DeclarativeBase):# 每条数据新增自动填充当前时间created_at:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,comment创建时间)# 新增默认时间数据修改时自动刷新为当前时间updated_at:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,onupdatedatetime.now,comment更新时间)#定义数据表news的模型类classNews(Base):#创建索引提高查询效率__table_args__(Index(fk_news_category_idx,category_id),#高频查询场景,外键列索引加速关联查询Index(idx_publish_time,publish_time),#按发布时间查询,发布时间索引加速排序/过滤)__tablename__news# 数据库表名# 新闻 ID主键自增id:Mapped[int]mapped_column(Integer,primary_keyTrue,autoincrementTrue,comment新闻ID)# 新闻标题非空长度 255title:Mapped[str]mapped_column(String(255),nullableFalse,comment新闻标题)# 新闻简介可选长度 500description:Mapped[Optional[str]]mapped_column(String(500),comment新闻简介)# 新闻内容文本类型非空content:Mapped[str]mapped_column(Text,nullableFalse,comment新闻内容)# 新闻图片 URL可选image:Mapped[Optional[str]]mapped_column(String(255),comment新闻图片)# 新闻作者可选长度 50author:Mapped[Optional[str]]mapped_column(String(50),comment新闻作者)# 新闻分类 ID外键关联 news_category 表的 ID非空category_id:Mapped[int]mapped_column(Integer,ForeignKey(news_category.id),nullableFalse,comment新闻分类ID)# 新闻浏览量默认 0非空views:Mapped[int]mapped_column(Integer,default0,nullableFalse,comment浏览量)# 发布时间默认当前时间非空publish_time:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,comment发布时间)def__repr__(self):returnfNews(id{self.id}, title{self.title}, views{self.views})
用python (fastapi)做项目第二天实现新闻列表和新闻详情接口
发布时间：2026/6/7 22:10:17
routers/news.py#routers接口路由层【接收前端参数、依赖注入数据库、调用 crud、组装返回格式】fromfastapiimportAPIRouter,Depends,HTTPException,Queryfromsqlalchemy.ext.asyncioimportAsyncSessionfromtypingimportListfrompydanticimportBaseModelfromconfig.db_confimportget_db# 获取数据库会话的依赖函数fromcrudimportnewsascrud_news# 导入crud数据库操作fromschemas.newsimport(CategoryCreate,CategoryUpdate,CategoryResponse)# 导入Pydantic出参模型# 获取分类下的文章列表接口GET /api/news/listrouter.get(/list)#alias起的别名asyncdefgetNewsList(category_id:intQuery(...,aliascategoryId),# 必填查询参数前端用 categoryIdpage:int1,# 页码默认 1page_size:intQuery(10,aliaspageSize,le100),# 每页条数默认 10最大 100db:AsyncSessionDepends(get_db)# 依赖注入数据库会话):获取新闻列表分页##思路:处理分页规则查询新闻列表 计算总量 计算是否还有更多try:# 计算跳过记录数偏移量skip(page-1)*page_size# 调用 CRUD 获取当前页的新闻列表返回 ORM 对象列表news_listawaitcrud_news.get_news_list(db,category_id,skip,page_size)# 调用 CRUD 获取该分类下的新闻总条数totalawaitcrud_news.get_news_count(db,category_id)# 判断是否还有下一页数据has_more(skiplen(news_list))totalreturn{code:200,message:获取新闻列表成功,data:{total:total,has_more:has_more,list:news_list# list: [NewsResponse.model_validate(news) for news in news_list]}}exceptExceptionase:# 可以根据需要记录日志raiseHTTPException(status_code500,detailf获取新闻列表失败:{str(e)})# 文章详情接口GET /api/news/detailrouter.get(/detail)asyncdefgetNewsDetail(id:intQuery(...,aliasid),db:AsyncSessionDepends(get_db)):获取新闻详情try:# 调用 CRUD 获取新闻详情newsawaitcrud_news.get_news_detail(db,id)ifnotnews:raiseHTTPException(status_code404,detail新闻不存在)#调用 CRUD 获取新闻浏览量news.viewsawaitcrud_news.get_news_views(db,news.id)awaitdb.refresh(news)# if not news.views:# raise HTTPException(status_code404, detail获取新闻浏览量失败)#调用 CRUD 获取同类的相关推介新闻news.related_newsawaitcrud_news.get_related_news(db,news.id,news.category_id)return{code:200,message:获取新闻详情成功,data:news}exceptExceptionase:# 可以根据需要记录日志raiseHTTPException(status_code500,detailf获取新闻详情失败:{str(e)})crud/news.pyfromsqlalchemy.ext.asyncioimportAsyncSessionfromsqlalchemyimportselect,func,updatefromsqlalchemy.ormimportselectinloadfromtypingimportList,Optionalfrommodels.newsimportCategory,News# 导入数据库映射模型fromschemas.newsimportCategoryCreate,CategoryUpdate#查询ID分类下的文章列表asyncdefget_news_list(db:AsyncSession,category_id:int,skip:int0,limit:int10)-List[News]:获取分类下的文章列表分页# # 构建查询语句SELECT * FROM news WHERE category_id :id OFFSET :skip LIMIT :limitstmtselect(News).where(News.category_idcategory_id).offset(skip).limit(limit)#缺少 order_by结果顺序不确定建议加上 .order_by(News.publish_time.desc())resultawaitdb.execute(stmt)# 提取所有标量News 对象返回列表returnresult.scalars().all()# 实际返回 List[News]#当前ID下的新闻数量asyncdefget_news_count(db:AsyncSession,category_id:int)-int:获取分类下的文章数量#func.count()聚合计算# # 构建计数查询SELECT COUNT(news.id) FROM news WHERE category_id :idstmtselect(func.count(News.id)).where(News.category_idcategory_id)resultawaitdb.execute(stmt)returnresult.scalar_one()# scalar_one()返回单个结果 scalar_one_or_none()返回单个结果或None#查询文章详情asyncdefget_news_detail(db:AsyncSession,news_id:int):获取文章详情stmtselect(News).where(News.idnews_id)resultawaitdb.execute(stmt)returnresult.scalar_one_or_none()# 查询文章浏览量asyncdefget_news_views(db:AsyncSession,news_id:int):获取文章浏览量update_stmtupdate(News).where(News.idnews_id).values(viewsNews.views1)awaitdb.execute(update_stmt)# await db.commit()# #更新检查数据库是否真的命中了数据 命中了返回True# return result.rowcount 0#获取同类的相关推介新闻asyncdefget_related_news(db:AsyncSession,news_id:int,category_id:int,limit:int10):获取同类新闻#order_by 排序 浏览量和发布时间 News.views.desc()默认是升序stmtselect(News).where(News.category_idcategory_id).where(News.id!news_id).order_by(News.views.desc(),News.publish_time.desc()).limit(limit)resultawaitdb.execute(stmt)returnresult.scalars().all()models/news.pyfromdatetimeimportdatetimefromsqlalchemyimportDateTime,String,Integer,Text,ForeignKeyfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnfromsqlalchemy.schemaimportIndexfromtypingimportOptional## 所有数据库表的父类公共字段抽离创建时间、更新时间#Base抽取公共时间字段所有数据表继承后自动带上创建 / 更新时间不用重复写classBase(DeclarativeBase):# 每条数据新增自动填充当前时间created_at:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,comment创建时间)# 新增默认时间数据修改时自动刷新为当前时间updated_at:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,onupdatedatetime.now,comment更新时间)#定义数据表news的模型类classNews(Base):#创建索引提高查询效率__table_args__(Index(fk_news_category_idx,category_id),#高频查询场景,外键列索引加速关联查询Index(idx_publish_time,publish_time),#按发布时间查询,发布时间索引加速排序/过滤)__tablename__news# 数据库表名# 新闻 ID主键自增id:Mapped[int]mapped_column(Integer,primary_keyTrue,autoincrementTrue,comment新闻ID)# 新闻标题非空长度 255title:Mapped[str]mapped_column(String(255),nullableFalse,comment新闻标题)# 新闻简介可选长度 500description:Mapped[Optional[str]]mapped_column(String(500),comment新闻简介)# 新闻内容文本类型非空content:Mapped[str]mapped_column(Text,nullableFalse,comment新闻内容)# 新闻图片 URL可选image:Mapped[Optional[str]]mapped_column(String(255),comment新闻图片)# 新闻作者可选长度 50author:Mapped[Optional[str]]mapped_column(String(50),comment新闻作者)# 新闻分类 ID外键关联 news_category 表的 ID非空category_id:Mapped[int]mapped_column(Integer,ForeignKey(news_category.id),nullableFalse,comment新闻分类ID)# 新闻浏览量默认 0非空views:Mapped[int]mapped_column(Integer,default0,nullableFalse,comment浏览量)# 发布时间默认当前时间非空publish_time:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,comment发布时间)def__repr__(self):returnfNews(id{self.id}, title{self.title}, views{self.views})
