如何序列化SQLAlchemy关联查询结果为JSON？

已知有疑似重复问题，但现有解决方案无效。需求是将SQLAlchemy的Join Query结果转换为JSON，要求输出包含关联双方表的所有列，且完整体现一对多关系。目前找到的方案仅能序列化单表列，或仅能体现单一关联而非完整的一对多结构。

最小复现示例 #

models.py

from dataclasses import dataclass
from typing import List

from sqlalchemy import create_engine, Column, ForeignKey, Integer, String
from sqlalchemy.orm import declarative_base, relationship, sessionmaker, Mapped

Base = declarative_base()

class BaseSerializable(Base):
    __abstract__ = True
    def as_dict(self):
       return {c.name: getattr(self, c.name) for c in self.__table__.columns}

@dataclass
class Player(BaseSerializable):
    __tablename__ = "customers"

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    scores: Mapped[List["Score"]] = relationship()

@dataclass
class Score(BaseSerializable):
    __tablename__ = "scores"

    id = Column(Integer, primary_key=True)
    player_id: Mapped[int] = Column(Integer, ForeignKey("customers.id"))
    value = Column(Integer)

engine = create_engine('sqlite:///database.db')
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)
db = SessionLocal()

main.py

from json import dumps, JSONEncoder
from fastapi.encoders import jsonable_encoder
from models import db, Player, Score
from sqlalchemy import Row
from sqlalchemy.orm import DeclarativeMeta

db.query(Player).delete()
db.query(Score).delete()
db.add(
    Player(
        name="Test Player 1"
    )
)

db.add(
    Score(
        player_id=1,
        value=9001
    )
)
db.add(
    Score(
        player_id=1,
        value=9002
    )
)
db.commit()


player: Player = (
    db.query(Player)
        .filter(Player.id == 1)
        .join(Score)
        .first()
)

# FastAPI编码器
def fastapi_json_encode(o):
    return dumps(jsonable_encoder(o))

# 自定义SQLAlchemy编码器
class SqlAlchemyEncoder(JSONEncoder):
    def default(self, o):
        if isinstance(o, Row):
            data = {}
            for obj in o:
                data.update(self.parse_sqlalchemy_object(obj))
            return data
        if isinstance(o.__class__, DeclarativeMeta):
            return self.parse_sqlalchemy_object(o)
        return JSONEncoder.default(self, o)


    def parse_sqlalchemy_object(self, o):
        data = {}
        fields = o.__json__() if hasattr(o, '__json__') else dir(o)
        for field in [f for f in fields if not f.startswith('_') and f not in ['metadata', 'query', 'query_class', 'registry']]:
            value = o.__getattribute__(field)
            try:
                dumps(value)
                data[field] = value
            except TypeError:
                data[field] = None
        return data

print(f'{SqlAlchemyEncoder().default(player)=}')
print(f'{fastapi_json_encode(player)=}')
print(f'{player.as_dict()=}')

预期与实际输出 #

预期输出应为类似如下结构：

{"id": 1, "name": "Test Player 1", "scores": [{"id": 1, "value": 9001}, {"id": 2, "value": 9002}]}

实际运行输出：

SqlAlchemyEncoder().default(player)={'as_dict': None, 'id': 1, 'name': 'Test Player 1', 'scores': None}
fastapi_json_encode(player)='{"scores": [{"player_id": 1}, {"player_id": 1}]}'
player.as_dict()={'id': 1, 'name': 'Test Player 1'}

尝试指定多类查询的结果 #

修改查询语句为同时指定Player和Score：

player: Player = (
    db.query(Player, Score)
        .filter(Player.id == 1)
        .join(Score)
        .first()
)

仅保留自定义编码器的输出，结果为：

SqlAlchemyEncoder().default(player)={'as_dict': None, 'id': 1, 'name': 'Test Player 1', 'scores': None, 'player_id': 1, 'value': 9001}

此结果仅返回关联查询第一行的序列化数据，而非包含所有关联Score的Player完整结构，不符合需求。

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