FastAPI+SQLAlchemy ORM获取关联表字段的正确方法及报错解决

方法一：直接查询所需字段 #

不需要查询整个关联表模型，只把需要的城市、州、国家名称字段加入查询，这样返回的结果可以直接处理成目标结构，避免Row对象的序列化问题：

query = (
    db.query(
        UserModel,
        CityModel.name.label("city_name"),
        StateModel.name.label("state_name"),
        CountryModel.name.label("country_name")
    )
    .join(UserDocumentModel)
    .join(UserAddressModel, UserAddressModel.user_id == UserModel.id)
    .join(CityModel, CityModel.id == UserAddressModel.city_id)
    .join(StateModel, StateModel.id == CityModel.state_id)
    .join(CountryModel, CountryModel.id == StateModel.country_id)
    .filter(
        UserModel.is_deleted == is_deleted,
        UserAddressModel.is_deleted == 0
    )
)

results = query.all()

拿到结果后，每个元素是(UserModel实例, 城市名, 州名, 国家名)的元组，可手动组装成所需结构：

user_list = []
for user, city_name, state_name, country_name in results:
    user_data = {k: v for k, v in user.__dict__.items() if not k.startswith("_")}
    user_data.update({
        "city_name": city_name,
        "state_name": state_name,
        "country_name": country_name
    })
    user_list.append(user_data)

如果用Pydantic，也可以直接定义包含这些字段的模型来接收转换。

方法二：利用ORM关系预加载 #

在模型中定义好关联关系，查询时通过joinedload预加载关联数据，直接通过UserModel的属性访问关联名称，代码更简洁：

第一步：定义模型间的关联关系 #

# UserModel
class UserModel(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    is_deleted = Column(Integer, default=0)
    # 假设用户与地址为一对一关系
    address = relationship("UserAddressModel", uselist=False, back_populates="user")

# UserAddressModel
class UserAddressModel(Base):
    __tablename__ = "user_addresses"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    city_id = Column(Integer, ForeignKey("cities.id"))
    is_deleted = Column(Integer, default=0)
    user = relationship("UserModel", back_populates="address")
    city = relationship("CityModel", back_populates="addresses")

# CityModel
class CityModel(Base):
    __tablename__ = "cities"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    state_id = Column(Integer, ForeignKey("states.id"))
    addresses = relationship("UserAddressModel", back_populates="city")
    state = relationship("StateModel", back_populates="cities")

# StateModel
class StateModel(Base):
    __tablename__ = "states"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    country_id = Column(Integer, ForeignKey("countries.id"))
    cities = relationship("CityModel", back_populates="state")
    country = relationship("CountryModel", back_populates="states")

# CountryModel
class CountryModel(Base):
    __tablename__ = "countries"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    states = relationship("StateModel", back_populates="country")

第二步：查询时预加载关联数据 #

用joinedload一次性加载所有关联层级，避免N+1查询问题：

from sqlalchemy.orm import joinedload

query = (
    db.query(UserModel)
    .join(UserDocumentModel)
    .options(
        joinedload(UserModel.address)
        .joinedload(UserAddressModel.city)
        .joinedload(CityModel.state)
        .joinedload(StateModel.country)
    )
    .filter(
        UserModel.is_deleted == is_deleted,
        UserAddressModel.is_deleted == 0
    )
)

results = query.all()

现在可直接通过实例属性获取名称：

for user in results:
    city_name = user.address.city.name
    state_name = user.address.city.state.name
    country_name = user.address.city.state.country.name

这种方式符合ORM使用习惯，返回的都是Model实例，序列化到Pydantic也更方便。

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