Skip to content

新闻分类

新闻分类获取

I.请求与响应过程

当用户访问首页时，首先需要从从后端获取新闻数据，然后前端通过js代码将获取的数据加载到页面中

flowchart  LR

    subgraph B [后端]
        U[FastAPI服务]
    end

    subgraph A [首页界面]
        direction LR
        F1[新闻分类]
        F2[新闻列表]
    end

   A -- ① 请求分类<br> /api/news/categories --> B
   B -- ② 返回响应 <br> {code: 200,message: success,data: [{id: 1,name: 头条}]}  -->  A

II. 路由定义

在routers目录下创建一个news.py文件，编写路由解析代码

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from db_conf import get_db



router = APIRouter(
    prefix="/api/news",  # 设置路由前缀，该路由器下的所有路径都会自动加上此前缀
    tags=["news"]  # 设置API标签，用于在API文档中对相关接口进行分组 可选
)


@router.get("/categories")
async def read_categories(skip: int = 0, limit: int = 100, db: AsyncSession = Depends(get_db)):
    """
        获取新闻分类列表接口
        Args:
            skip (int, optional): 跳过的记录数，用于分页，默认为0
            limit (int, optional): 返回的记录数限制，用于分页，默认为100
            db (AsyncSession): 通过依赖注入获取的数据库会话对象
        Example:
            GET /api/news/categories
            GET /api/news/categories?skip=0&limit=10
    """

    return  {
        "code": 200,
        "message": "success",
        "data": '新闻分类列表'
    }

image-20251025094639357

在main.py文件中进行路由注册

1
2
3
from routers import news
# ....其他代码....
app.include_router(news.router)

image-20251025095141378

重启FastAPI服务，运行测试，检查定义的路由接口是否能够处理前端发送的请求

image-20251025095513244

III. 定义模型类

定义对应的模型类，方便对应数据库中的数据进行ORM操作

在models目录下创建news.py文件，编写模型类

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer, DateTime
from datetime import datetime



class Base(DeclarativeBase):
    pass


class Category(Base):
    """
    新闻分类模型
    对应数据库中的 news_category 表
    """
    __tablename__ = 'news_category'

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, comment="分类ID")
    name: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, comment="分类名称")
    sort_order: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="排序顺序")
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now, comment="创建时间")
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.now, onupdate=datetime.now,
                                                 comment="更新时间")

    def __repr__(self):
        return f"<Category(id={self.id}, name='{self.name}', sort_order={self.sort_order})>"

image-20251025100419139

IV. 编写ORM数据操作

在curd目录下创建news.py文件，将新闻模块的数据操作按照对应的逻辑编写代码

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import List

from models.news import Category


async def get_categories(db: AsyncSession, skip: int = 0, limit: int = 100) -> List[Category]:
    """
        获取新闻分类列表，支持分页

    Args:
        db (AsyncSession): SQLAlchemy异步数据库会话对象
        skip (int, optional): 跳过的记录数，用于分页，默认为0
        limit (int, optional): 返回的记录数限制，用于分页，默认为100

    Returns:
        List[Category]: 返回分类对象列表，每个元素都是Category模型实例

    Example:
        # 获取前100个分类
        categories = await get_categories(db, skip=0, limit=100)

        # 获取第2页分类（每页10条）
        second_page_categories = await get_categories(db, skip=10, limit=10)

    """
    stmt = select(Category).offset(skip).limit(limit)
    result = await db.execute(stmt)
    return result.scalars().all()

image-20251025100807064

V. 路由中实现数据操作

回到对应的路由中，调用刚才编写好的数据处理函数。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
from curd import news

@router.get("/categories")
async def read_categories(skip: int = 0, limit: int = 100, db: AsyncSession = Depends(get_db)):
    """
        获取新闻分类列表接口
        Args:
            skip (int, optional): 跳过的记录数，用于分页，默认为0
            limit (int, optional): 返回的记录数限制，用于分页，默认为100
            db (AsyncSession): 通过依赖注入获取的数据库会话对象
        Example:
            GET /api/news/categories
            GET /api/news/categories?skip=0&limit=10
    """

    categories = await news.get_categories(db, skip=skip, limit=limit)

    return  {
        "code": 200,
        "message": "success",
        "data": categories
    }

image-20251025105906796

重启FastAPI服务，测试数据是否正常返回

image-20251025110012325

启动前端服务检查数据是否正常加载

image-20251025110438790

整体开发流程总结

  1. 定义路由匹配
  2. 路由注册
  3. 定义模型类
  4. 定义ORM数据操作逻辑
  5. 路由中调用数据逻辑函数