能否让SQLAlchemy关联查询返回自定义扩展类对象？

项目结构 #

app/
├─ classes/
│  ├─ person.py
│  ├─ address.py
├─ database.py
├─ main.py
├─ models.py

各文件代码 #

database.py #

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker, scoped_session

Base = declarative_base()
engine = create_engine('sqlite://', pool_pre_ping=True)

session_factory = sessionmaker(bind=engine)
ScopedSession = scoped_session(session_factory)


class Repr:
    def __repr__(self) -> str:
        # custom implementation of the __repr__ method
        print('some text')

models.py #

# models.py
from database import Base, Repr, engine
from sqlalchemy import Column, ForeignKey, Integer, String, Text
from sqlalchemy.orm import relationship, backref


class PersonDB(Base, Repr):
    __tablename__ = 'persons'
    id = Column(Integer, primary_key=True, nullable=False)
    address_id = Column(Integer, ForeignKey('address.id'), nullable=False)
    first_name = Column(String(45))
    last_name = Column(String(45))
    # relationships
    address = relationship('AddressDB', backref=backref('persons'))


class AddressDB(Base, Repr):
    __tablename__ = 'address'
    id = Column(Integer, primary_key=True, nullable=False)
    street = Column(Text)
    city = Column(Text)


Base.metadata.create_all(engine)

classes/person.py #

# person.py
from models import PersonDB


class Person(PersonDB):
    @property
    def full_name(self) -> str:
        return f'{self.first_name} {self.last_name}'

classes/address.py #

# address.py
from models import AddressDB


class Address(AddressDB):
    # no changes to the base class required.
    # The class merely exist as an alias importable from classes.
    pass 

核心问题 #

当前方案实现了扩展逻辑与表结构的分离，但通过SQLAlchemy的relationship获取关联对象时，返回的是基类（如PersonDB）而非扩展类（Person），导致无法调用扩展属性：

for p in a.persons:
    print(p.full_name)  # Error: p is PersonDB not Person object

需要解决的问题：如何让关联查询无缝返回扩展类实例，无需手动转换？或者是否需要调整现有架构？

方案一：重构继承关系，让扩展类直接作为映射类 #

将表结构定义直接迁移到classes文件夹的扩展类中，让扩展类成为SQLAlchemy的映射类，关联关系直接指向扩展类，从根源解决类型问题。

修改后文件内容：

• classes/person.py：

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


class Person(Base, Repr):
    __tablename__ = 'persons'
    id = Column(Integer, primary_key=True, nullable=False)
    address_id = Column(Integer, ForeignKey('address.id'), nullable=False)
    first_name = Column(String(45))
    last_name = Column(String(45))
    # 关联直接指向扩展类Address
    address = relationship('Address', backref=backref('persons'))

    @property
    def full_name(self) -> str:
        return f'{self.first_name} {self.last_name}'

• classes/address.py：

from database import Base, Repr
from sqlalchemy import Column, Integer, Text


class Address(Base, Repr):
    __tablename__ = 'address'
    id = Column(Integer, primary_key=True, nullable=False)
    street = Column(Text)
    city = Column(Text)

• models.py改为统一导入入口：

from classes.person import Person
from classes.address import Address

from database import engine, Base

Base.metadata.create_all(engine)

此方案下，a.persons返回的直接是Person类实例，可直接调用full_name属性，架构更清晰，无额外转换成本。

方案二：通过SQLAlchemy mapper重新绑定扩展类 #

如果不想迁移表结构定义，可通过SQLAlchemy底层的mapper将扩展类与数据库表重新关联，替换原基类的映射关系。

修改classes/person.py：

from sqlalchemy.orm import mapper
from models import PersonDB


class Person(PersonDB):
    @property
    def full_name(self) -> str:
        return f'{self.first_name} {self.last_name}'

# 将Person类绑定到persons表，继承原PersonDB的映射配置
mapper(Person, PersonDB.__table__, inherits=PersonDB.__mapper__)

修改classes/address.py：

from sqlalchemy.orm import mapper
from models import AddressDB


class Address(AddressDB):
    pass

mapper(Address, AddressDB.__table__, inherits=AddressDB.__mapper__)

修改models.py中的关联指向：

from database import Base, Repr, engine
from sqlalchemy import Column, ForeignKey, Integer, String, Text
from sqlalchemy.orm import relationship, backref


class PersonDB(Base, Repr):
    __tablename__ = 'persons'
    id = Column(Integer, primary_key=True, nullable=False)
    address_id = Column(Integer, ForeignKey('address.id'), nullable=False)
    first_name = Column(String(45))
    last_name = Column(String(45))
    # 关联指向扩展类Address
    address = relationship('Address')


class AddressDB(Base, Repr):
    __tablename__ = 'address'
    id = Column(Integer, primary_key=True, nullable=False)
    street = Column(Text)
    city = Column(Text)
    # 反向关联指向扩展类Person
    persons = relationship('Person')

# 导入扩展类触发映射绑定
from classes.person import Person
from classes.address import Address

Base.metadata.create_all(engine)

此方案无需迁移表结构，但需注意映射绑定的顺序，确保扩展类在基类映射后加载。

方案三：基类添加属性转换（妥协方案） #

如果无法修改现有架构，可在基类中添加属性转换逻辑，将关联查询结果转为扩展类实例，但此方式会创建新实例，可能脱离session管理，存在风险：

修改models.py的AddressDB：

from classes.person import Person

class AddressDB(Base, Repr):
    __tablename__ = 'address'
    id = Column(Integer, primary_key=True, nullable=False)
    street = Column(Text)
    city = Column(Text)
    
    @property
    def persons(self):
        # 将基类实例转为扩展类实例
        return [Person(**p.__dict__) for p in self._persons]
    
    # 原关联改名为_persons，作为内部逻辑
    _persons = relationship('PersonDB')

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