SQLAlchemy ORM中为反射属性构建column_property的最优方式

针对你遇到的反射模型下无法直接在column_property中引用ORM属性的问题，这里提供两种符合SQLAlchemy规范的纯ORM风格解决方案：

方案一：利用lambda延迟求值（纯反射场景首选） #

在类体定义column_property时，HeaderOrm类尚未完成初始化，直接引用id会触发未定义错误。通过lambda延迟表达式的求值时机，直到类创建完成后再解析，就能正常引用ORM的HeaderOrm.id属性：

from sqlalchemy import func, select
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    relationship,
    column_property,
)
from sqlalchemy.sql.schema import Table

class Base(DeclarativeBase):
    pass

class MessageOrm(Base):
    __table__ = Table(
        "single_recipient_message",
        Base.metadata,
        autoload_with=engine,
    )

    header: Mapped["HeaderOrm"] = relationship(
        back_populates="single_recipient_messages"
    )

class HeaderOrm(Base):
    __table__ = Table(
        "single_recipient_message_header",
        Base.metadata,
        autoload_with=engine,
    )

    single_recipient_messages: Mapped[list["MessageOrm"]] = relationship(
        back_populates="header"
    )
    
    msg_count = column_property(
        select(func.count(MessageOrm.id))
        .where(MessageOrm.header_id == lambda: HeaderOrm.id)  # 延迟引用ORM属性
        .correlate_except(MessageOrm)
        .scalar_subquery()
    )

方案二：显式声明模型列（适合需自定义列属性的场景） #

如果允许在反射基础上补充显式列声明，直接在类中定义id属性，就能在column_property中直接引用：

from sqlalchemy import Integer, func, select
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    relationship,
    column_property,
)
from sqlalchemy.sql.schema import Table

class Base(DeclarativeBase):
    pass

class MessageOrm(Base):
    __table__ = Table(
        "single_recipient_message",
        Base.metadata,
        autoload_with=engine,
    )

    header: Mapped["HeaderOrm"] = relationship(
        back_populates="single_recipient_messages"
    )

class HeaderOrm(Base):
    __table__ = Table(
        "single_recipient_message_header",
        Base.metadata,
        autoload_with=engine,
    )
    
    # 显式声明主键列，与反射表列映射
    id: Mapped[int] = mapped_column(Integer, primary_key=True)

    single_recipient_messages: Mapped[list["MessageOrm"]] = relationship(
        back_populates="header"
    )
    
    msg_count = column_property(
        select(func.count(MessageOrm.id))
        .where(MessageOrm.header_id == id)
        .correlate_except(MessageOrm)
        .scalar_subquery()
    )

补充说明 #

使用__table__.c.id本身并非不规范，SQLAlchemy官方允许Core与ORM语法混合使用，但如果追求纯ORM风格，上述两种方案更贴合需求：

• 方案一无需修改反射逻辑，仅通过延迟求值解决类初始化顺序问题；
• 方案二通过显式列声明让代码更直观，同时保留ORM操作方式。

内容的提问来源于stack exchange，提问作者Pierre Massé