FastAPI中SQLAlchemy联表查询返回对象序列化报错求助

在学习FastAPI时，通过SQLAlchemy关联Post和Vote表统计每篇帖子的投票数，能得到投票总数，但返回结果时Post以对象实例形式存在而非实际数据，触发500内部服务器错误。

模型代码 #

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, nullable=False)
    title = Column(String, nullable=False)
    content = Column(String, nullable=False)
    published = Column(Boolean, server_default="TRUE")
    created_at = Column(TIMESTAMP(timezone=True),
                        nullable=False, server_default=text('now()'))
    owner_id = Column(Integer, ForeignKey(
        "users.id", ondelete="CASCADE"), nullable=False)

    owner = relationship("User")


class Vote(Base):
    __tablename__ = "votes"

    user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), primary_key=True)
    post_id = Column(Integer, ForeignKey("posts.id", ondelete="CASCADE"), primary_key=True)
    created_at = Column(TIMESTAMP(timezone=True),
                        nullable=False, server_default=text('now()'))

路由函数代码 #

@router.get("/posts")
async def get_posts(
        db: Session = Depends(get_db),
        current_user: int = Depends(oauth2.get_current_user),
        search: Optional[str] = '',
):

    posts = db.query(models.Post).filter(models.Post.title.contains(search)).all()


    results = db.query(models.Post, func.count(models.Vote.post_id).label("votes_count")).\
join(models.Vote, models.Vote.post_id == models.Post.id, isouter=True).\
group_by(models.Post.id).all()

print(results)


    return results

打印的results输出 #

[(<app.models.Post object at 0x7f19893fe310>, 0), (<app.models.Post object at 0x7f19893fe390>, 0), (<app.models.Post object at 0x7f19893fe210>, 1), (<app.models.Post object at 0x7f19893fe290>, 0), (<app.models.Post object at 0x7f19893fe410>, 0)]

报错信息 #

INFO:     127.0.0.1:50202 - "GET /posts HTTP/1.1" 500 Internal Server Error
ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/fastapi/encoders.py", line 322, in jsonable_encoder
    data = dict(obj)
           ^^^^^^^^^
TypeError: cannot convert dictionary update sequence element #0 to a sequence

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/fastapi/encoders.py", line 327, in jsonable_encoder
    data = vars(obj)
           ^^^^^^^^^
TypeError: vars() argument must have __dict__ attribute

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/uvicorn/protocols/http/httptools_impl.py", line 426, in run_asgi
    result = await app(  # type: ignore[func-returns-value]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/uvicorn/middleware/proxy_headers.py", line 84, in __call__
    return await self.app(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/fastapi/applications.py", line 1106, in __call__
    await super().__call__(scope, receive, send)
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/starlette/applications.py", line 122, in __call__
    await self.middleware_stack(scope, receive, send)
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/starlette/middleware/errors.py", line 184, in __call__
    raise exc
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/starlette/middleware/errors.py", line 162, in __call__
    await self.app(scope, receive, _send)
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 79, in __call__
    raise exc
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 68, in __call__
    await self.app(scope, receive, sender)
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 20, in __call__
    raise e
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 17, in __call__
    await self.app(scope, receive, send)
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/starlette/routing.py", line 718, in __call__
    await route.handle(scope, receive, send)
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/starlette/routing.py", line 276, in handle
    await self.app(scope, receive, send)
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/starlette/routing.py", line 66, in app
    response = await func(request)
               ^^^^^^^^^^^^^^^^^^^
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/fastapi/routing.py", line 292, in app
    content = await serialize_response(
              ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/fastapi/routing.py", line 180, in serialize_response
    return jsonable_encoder(response_content)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/fastapi/encoders.py", line 301, in jsonable_encoder
    jsonable_encoder(
  File "/home/eminent/Documents/My codes/FastAPI/Tutorials/SocialAPI/venv/lib/python3.11/site-packages/fastapi/encoders.py", line 330, in jsonable_encoder
    raise ValueError(errors) from e
ValueError: [TypeError('cannot convert dictionary update sequence element #0 to a sequence'), TypeError('vars() argument must have __dict__ attribute')]      

核心问题是FastAPI无法直接序列化SQLAlchemy的ORM对象，需要将查询结果转换为可序列化的字典结构，同时保留投票数。

方法一：使用Pydantic模型转换 #

1. 定义包含投票数的Pydantic响应模型：

from pydantic import BaseModel
from datetime import datetime
from typing import List

class UserOut(BaseModel):
    id: int
    # 补充User模型的其他字段，比如username、email等
    class Config:
        orm_mode = True

class PostOut(BaseModel):
    id: int
    title: str
    content: str
    published: bool
    created_at: datetime
    owner_id: int
    owner: UserOut
    votes_count: int

    class Config:
        orm_mode = True

2. 修改路由函数，转换查询结果：

@router.get("/posts", response_model=List[PostOut])
async def get_posts(
        db: Session = Depends(get_db),
        current_user: int = Depends(oauth2.get_current_user),
        search: Optional[str] = '',
):
    results = db.query(models.Post, func.count(models.Vote.post_id).label("votes_count")).\
        join(models.Vote, models.Vote.post_id == models.Post.id, isouter=True).\
        filter(models.Post.title.contains(search)).\
        group_by(models.Post.id).all()

    # 转换结果为符合Pydantic模型的字典列表
    response = []
    for post, votes_count in results:
        post_dict = post.__dict__
        post_dict.pop('_sa_instance_state', None)  # 移除SQLAlchemy内部状态字段
        post_dict['votes_count'] = votes_count
        response.append(post_dict)
    
    return response

方法二：直接构造字典返回 #

如果不需要Pydantic验证，可直接手动构造包含所有字段的字典：

@router.get("/posts")
async def get_posts(
        db: Session = Depends(get_db),
        current_user: int = Depends(oauth2.get_current_user),
        search: Optional[str] = '',
):
    results = db.query(models.Post, func.count(models.Vote.post_id).label("votes_count")).\
        join(models.Vote, models.Vote.post_id == models.Post.id, isouter=True).\
        filter(models.Post.title.contains(search)).\
        group_by(models.Post.id).all()

    return [
        {
            "id": post.id,
            "title": post.title,
            "content": post.content,
            "published": post.published,
            "created_at": post.created_at,
            "owner_id": post.owner_id,
            "owner": {
                "id": post.owner.id,
                # 补充owner的其他字段，如username、email
            },
            "votes_count": votes_count
        }
        for post, votes_count in results
    ]

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