FastAPI中如何实现类似Django SerializerMethodField的功能？

FastAPI中没有Django Serializer的context机制，但可以通过依赖注入获取当前用户，结合Pydantic模型动态生成包含点赞状态的响应数据，逻辑和你Django中的实现完全对应：

核心思路 #

1. 通过依赖注入获取当前登录用户（或处理匿名场景）
2. 查询目标评论列表后，逐个判断用户是否点赞该评论
3. 将点赞状态添加到最终的响应数据中

代码实现 #

1. 定义数据库模型（SQLAlchemy示例） #

对应你Django中的ORM模型结构：

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from database import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    uuid = Column(String, unique=True, index=True)
    comment_favors = relationship("CommentFavorRecord", back_populates="user")

class CommentRecord(Base):
    __tablename__ = "comments"
    id = Column(Integer, primary_key=True, index=True)
    content = Column(String)
    depth = Column(Integer)
    # 其他字段：user, article, root, replyId...
    favors = relationship("CommentFavorRecord", back_populates="comment")

class CommentFavorRecord(Base):
    __tablename__ = "comment_favors"
    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    comment_id = Column(Integer, ForeignKey("comments.id"))
    user = relationship("User", back_populates="comment_favors")
    comment = relationship("CommentRecord", back_populates="favors")

2. 实现获取当前用户的依赖 #

通过依赖注入获取登录用户（这里以JWT认证为例，可根据你的实际认证方式调整）：

from fastapi import Depends, HTTPException, status
from jose import JWTError, jwt
from sqlalchemy.orm import Session
from database import get_db
from models import User
from fastapi.security import OAuth2PasswordBearer

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(
    db: Session = Depends(get_db),
    token: str = Depends(oauth2_scheme)
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭证",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_uuid: str = payload.get("sub")
        if user_uuid is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = db.query(User).filter(User.uuid == user_uuid).first()
    if user is None:
        raise credentials_exception
    return user

3. 定义Pydantic响应模型 #

创建包含favor字段的响应模型，继承基础评论模型：

from pydantic import BaseModel
from typing import Optional

class CommentBase(BaseModel):
    id: int
    content: str
    depth: int
    # 其他需要返回的字段：user, article, root, replyId...

    class Config:
        from_attributes = True  # Pydantic v2 用于从ORM模型转换

class CommentResponse(CommentBase):
    favor: bool  # 新增点赞状态字段

4. 路由处理逻辑 #

在路由中获取当前用户，查询评论并动态计算点赞状态：

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from database import get_db
from models import CommentRecord, CommentFavorRecord
from schemas import CommentResponse
from dependencies import get_current_user
from models import User
from typing import Optional

router = APIRouter(prefix="/comments", tags=["评论"])

@router.get("/level2/", response_model=list[CommentResponse])
def get_level2_comments(
    db: Session = Depends(get_db),
    current_user: Optional[User] = Depends(get_current_user)
):
    # 查询二级评论（根据你的业务条件调整）
    comments = db.query(CommentRecord).filter(CommentRecord.depth == 2).all()
    
    comment_responses = []
    for comment in comments:
        # 判断当前用户是否点赞
        favor = False
        if current_user:
            # 检查是否存在点赞记录
            has_favor = db.query(CommentFavorRecord).filter(
                CommentFavorRecord.comment_id == comment.id,
                CommentFavorRecord.user_id == current_user.id
            ).first() is not None
            favor = has_favor
        
        # 生成响应对象：从ORM模型转换后，更新favor字段
        comment_data = CommentResponse.from_orm(comment).model_copy(update={"favor": favor})
        comment_responses.append(comment_data)
    
    return comment_responses

优化点 #

可以把判断点赞的逻辑封装成独立函数，提升代码可读性：

def check_comment_favor(db: Session, comment: CommentRecord, user: Optional[User]) -> bool:
    if not user:
        return False
    # 使用exists()提升查询性能
    return db.query(CommentFavorRecord).filter(
        CommentFavorRecord.comment == comment,
        CommentFavorRecord.user == user
    ).exists().scalar()

然后在路由中调用：

for comment in comments:
    favor = check_comment_favor(db, comment, current_user)
    comment_responses.append(CommentResponse.from_orm(comment).model_copy(update={"favor": favor}))

内容的提问来源于stack exchange，提问作者Andy Hsieh