收藏查询

查询收藏状态

I.请求与响应过程

当用户进入详情页后，需要先请求后端，获取当前新闻的收藏状态

flowchart  LR

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

    subgraph A [用户界面]
        direction LR
        F1[新闻详情]
    end

   A -- ① 请求分类<br> /api/favorite/check --> B
   B -- ② 返回响应 <br> {'code': 200,'message': 'success','data':{ 'isFavorite': true  }}  -->  A

II. 数据校验模型

当前业务中，检查收藏数据无需校验

III. 路由定义

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

from fastapi import APIRouter, Depends, Header,Query
from sqlalchemy.ext.asyncio import AsyncSession
from db_conf import get_db



router = APIRouter(
    prefix="/api/favorite",
    tags=["favorite"]
)


@router.get("/check")
async def check_favorite_status(
        news_id: int=Query( ...,alias='newsId'),
        authorization: str = Header(...),
        db: AsyncSession = Depends(get_db)
):
    """
    检查用户是否已收藏指定新闻

    Args:
        news_id: 新闻ID
        authorization: Authorization请求头(用户令牌)
        db: 数据库会话

    Returns:
        dict: 包含收藏状态的响应
    """

    # 返回检查结果
    return {
        "code": 200,
        "message": "success",
        "data": '是否收藏'
    }

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

from routers import news,users,favorite

# ....其他代码....

app.include_router(favorite.router)

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

### 收藏状态查询
GET http://127.0.0.1:8000//api/favorite/check?newsId=1
Content-Type: application/json
Authorization: 597d92a4-2183-49e5-b14e-04fa95b6f6b8

IV. 定义模型类

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

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

from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase
from sqlalchemy import Integer, DateTime, ForeignKey, Index, UniqueConstraint
from datetime import datetime
from .users import User
from .news import News


class Base(DeclarativeBase):
    pass


class Favorite(Base):
    """
    收藏表ORM模型
    """
    __tablename__ = 'favorite'

    # 创建索引
    __table_args__ = (
        UniqueConstraint('user_id', 'news_id', name='user_news_unique'),
        Index('fk_favorite_user_idx', 'user_id'),
        Index('fk_favorite_news_idx', 'news_id'),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True, comment="收藏ID")
    user_id: Mapped[int] = mapped_column(Integer, ForeignKey(User.id), nullable=False, comment="用户ID")
    news_id: Mapped[int] = mapped_column(Integer, ForeignKey(News.id), nullable=False, comment="新闻ID")
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, nullable=False, comment="收藏时间")

    def __repr__(self):
        return f"<Favorite(id={self.id}, user_id={self.user_id}, news_id={self.news_id}, created_at={self.created_at})>"

V. 编写ORM数据操作

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

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from models.favorite import Favorite



async def is_news_favorited(db: AsyncSession, user_id: int, news_id: int) -> bool:
    """
    检查用户是否已收藏某篇新闻

    Args:
        db: 数据库会话
        user_id: 用户ID
        news_id: 新闻ID

    Returns:
        bool: 是否已收藏
    """
    query = select(Favorite).where(
        Favorite.user_id == user_id,
        Favorite.news_id == news_id
    )
    result = await db.execute(query)
    return result.scalar_one_or_none() is not None

VI. 路由中实现数据操作

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

from fastapi import APIRouter, Depends, Header, Query, HTTPException
from starlette import status


@router.get("/check")
async def check_favorite_status(
        news_id: int=Query( ...,alias='newsId'),
        authorization: str = Header(...),
        db: AsyncSession = Depends(get_db)
):
    """
    检查用户是否已收藏指定新闻

    Args:
        news_id: 新闻ID
        authorization: Authorization请求头(用户令牌)
        db: 数据库会话

    Returns:
        dict: 包含收藏状态的响应
    """
    try:
        # 从Authorization头中提取token并验证用户
        token = authorization
        user = await users.get_user_by_token(db, token)
        if not user:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="无效的认证令牌或令牌已过期"
            )

        # 检查是否已收藏该新闻
        is_favorite = await favorite.is_news_favorited(db, user.id, news_id)

        # 返回检查结果
        return {
            "code": 200,
            "message": "success",
            "data": {
                "isFavorite": is_favorite
            }
        }

    except HTTPException:
        # 重新抛出自定义异常
        raise
    except Exception as e:
        # 处理其他异常
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="检查收藏状态失败，请稍后重试"
        )

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