SQLAlchemy序列化父子对象：懒加载相关返回异常问题

你调试时能看到User对象中已填充domainName子对象，说明SQLAlchemy的lazy=False已经生效，问题出在对象转JSON的序列化环节——默认的序列化工具不会自动识别并包含SQLAlchemy的relationship关联字段。

以下是几种可行的解决方法：

方法1：使用Pydantic模型（推荐，适配FastAPI等现代框架） #

通过定义嵌套的Pydantic Schema，显式声明要包含的关联字段，确保序列化时能完整输出数据：

from pydantic import BaseModel, ConfigDict
from typing import List

# 定义DomainName的序列化模型
class DomainNameSchema(BaseModel):
    domain_id: int
    user_id: str  # 匹配User表的主键类型（email是字符串）
    domain_name: str

    model_config = ConfigDict(from_attributes=True)  # 支持从SQLAlchemy对象导入数据

# 定义User的序列化模型，嵌套DomainNameSchema
class UserSchema(BaseModel):
    email: str
    username: str
    password: bytes  # BYTEA类型会自动序列化为base64字符串
    firstname: str
    lastname: str
    domainName: List[DomainNameSchema]  # 显式包含关联字段

    model_config = ConfigDict(from_attributes=True)

在服务中返回时，将SQLAlchemy对象转换为Pydantic模型：

user_obj = get_data(session, User, {"email": "r@tester.com"})
return UserSchema.model_validate(user_obj).model_dump()

方法2：在SQLAlchemy模型中添加自定义序列化方法 #

直接在User类中编写to_dict方法，手动处理关联字段的序列化：

class User(Base):
    __tablename__ = "user_account"

    # 现有字段定义...

    domainName = relationship("DomainName", cascade="all, delete-orphan", backref="user_account", lazy=False)

    def to_dict(self):
        return {
            "email": self.email,
            "username": self.username,
            "password": self.password.decode('utf-8') if self.password else "",
            "firstname": self.firstname,
            "lastname": self.lastname,
            "domainName": [
                {
                    "domain_id": d.domain_id,
                    "user_id": d.user_id,
                    "domain_name": d.domain_name
                } for d in self.domainName
            ]
        }

返回时调用该方法：

user_obj = get_data(session, User, {"email": "r@tester.com"})
return user_obj.to_dict()

方法3：自定义JSON编码器（适配标准json模块） #

如果直接使用Python标准库的json.dumps，需要自定义编码器来处理SQLAlchemy对象和关联字段：

import json
from sqlalchemy.ext.declarative import DeclarativeMeta

class AlchemyEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj.__class__, DeclarativeMeta):
            # 处理SQLAlchemy模型对象
            fields = {}
            for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
                data = obj.__getattribute__(field)
                try:
                    json.dumps(data)
                    fields[field] = data
                except TypeError:
                    # 递归处理关联的列表对象
                    if isinstance(data, list):
                        fields[field] = [self.default(item) for item in data]
            return fields
        return super().default(obj)

使用时：

user_obj = get_data(session, User, {"email": "r@tester.com"})
return json.dumps(user_obj, cls=AlchemyEncoder)

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