SQLAlchemy 2.0：会话管理、对象生命周期与高级查询技巧

from sqlalchemy import create_engine, Integer, String, DateTime, ForeignKey, func, select, desc
from sqlalchemy.orm import declarative_base, Session, Mapped, mapped_column, relationship, aliased

Base = declarative_base()

class Subject(Base):
    __tablename__ = 'subjects'

    id: Mapped[int] = mapped_column(primary_key=True)
    first_name: Mapped[str] = mapped_column(String(60), nullable=False)
    last_name: Mapped[str] = mapped_column(String(60), nullable=False)
    visits: Mapped[list['Visit']] = relationship(cascade='all, delete-orphan',
                                                 back_populates='subject')

    def __repr__(self):
        return f"<Subject(id={self.id}, name='{self.first_name} {self.last_name}')>"

class Visit(Base):
    __tablename__ = 'visits'

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    date: Mapped[DateTime] = mapped_column(DateTime, nullable=False)
    amount_spent: Mapped[int] = mapped_column(Integer, nullable=False)
    units: Mapped[str] = mapped_column(String, nullable=False)
    subject_id: Mapped[int] = mapped_column(Integer, ForeignKey('subjects.id'), index=True)
    subject: Mapped['Subject'] = relationship(back_populates='visits')

    def __repr__(self):
        # 注意：访问self.date必须在会话激活状态下，否则可能引发DetachedInstanceError
        # 更好的做法是在需要时才格式化，或确保对象处于“attached”状态
        try:
            return f"<Visit(id={self.id}, date='{self.date.strftime('%Y-%m-%d')}', subject_id={self.subject_id})>"
        except Exception:
            return f"<Visit(id={self.id}, date='[detached]', subject_id={self.subject_id})>"

# 数据库引擎配置 (这里使用SQLite内存数据库进行演示)
engine = create_engine('sqlite:///:memory:', echo=False)
Base.metadata.create_all(engine)

# 示例数据填充
with Session(engine) as session:
    subject1 = Subject(first_name="Alice", last_name="Smith")
    subject2 = Subject(first_name="Bob", last_name="Johnson")
    subject3 = Subject(first_name="Charlie", last_name="Brown")

    session.add_all([subject1, subject2, subject3])
    session.commit()

    session.add_all([
        Visit(subject=subject1, date=func.datetime('now', '-5 days'), amount_spent=100, units='USD'),
        Visit(subject=subject1, date=func.datetime('now', '-2 days'), amount_spent=120, units='USD'), # Alice's latest
        Visit(subject=subject2, date=func.datetime('now', '-7 days'), amount_spent=50, units='USD'),
        Visit(subject=subject2, date=func.datetime('now', '-1 day'), amount_spent=75, units='USD'),  # Bob's latest
        Visit(subject=subject3, date=func.datetime('now', '-3 days'), amount_spent=200, units='USD'), # Charlie's latest
    ])
    session.commit()

# 模拟错误情景
with Session(engine) as session:
    latest_visit = session.scalars(select(Visit).order_by(Visit.date.desc()).limit(1)).first()
    session.commit() # 此时 latest_visit 对象被标记为过期

# 在会话外部尝试访问过期对象的属性
try:
    print(f"Latest visit date (error expected): {latest_visit.date.strftime('%Y-%m-%d')}")
except Exception as e:
    print(f"捕获到错误: {type(e).__name__}: {e}")

with Session(engine) as session:
    latest_visit = session.scalars(select(Visit).order_by(Visit.date.desc()).limit(1)).first()
    if latest_visit:
        # 在会话内访问属性，此时对象是“attached”状态
        print(f"Latest visit date (within session): {latest_visit.date.strftime('%Y-%m-%d')}")
    session.commit() # 提交后，latest_visit 仍会过期

# 设置 expire_on_commit=False
with Session(engine, expire_on_commit=False) as session:
    latest_visit = session.scalars(select(Visit).order_by(Visit.date.desc()).limit(1)).first()
    session.commit() # 此时 latest_visit 不会过期

if latest_visit:
    # 在会话外部可以访问属性
    print(f"Latest visit date (expire_on_commit=False): {latest_visit.date.strftime('%Y-%m-%d')}")

with Session(engine) as session:
    # 使用 joinedload 预加载 subject
    stmt = select(Visit).options(relationship(Visit.subject)).order_by(Visit.date.desc()).limit(1)
    latest_visit = session.scalars(stmt).first()
    session.commit() # latest_visit 及其 subject 都会过期

if latest_visit:
    try:
        # 即使 visit 过期，如果 subject 之前被加载，其 scalar 属性可能仍然可访问
        # 但如果 subject 也是过期状态，访问其属性仍会出错
        print(f"Latest visit by {latest_visit.subject.first_name} (DetachedInstanceError may still occur if subject is expired)")
    except Exception as e:
        print(f"捕获到错误 (预加载后访问过期关系): {type(e).__name__}: {e}")

# 如果需要访问关系属性，通常也需要在会话内处理，或者使用 expire_on_commit=False
with Session(engine, expire_on_commit=False) as session:
    stmt = select(Visit).options(relationship(Visit.subject)).order_by(Visit.date.desc()).limit(1)
    latest_visit = session.scalars(stmt).first()
    session.commit()

if latest_visit:
    print(f"Latest visit by {latest_visit.subject.first_name} (expire_on_commit=False with joinedload)")

with Session(engine) as session:
    # 1. 构建一个子查询，使用 ROW_NUMBER() 为每个 subject_id 分组内的 visit 记录按日期降序编号
    #    partition_by=Visit.subject_id 定义了分组
    #    order_by=Visit.date.desc() 定义了组内排序规则
    subquery = (
        select(
            Visit,
            func.row_number()
            .over(partition_by=Visit.subject_id, order_by=Visit.date.desc())
            .label("rn"), # 给行号起一个别名 'rn'
        )
        .subquery() # 将此查询包装成一个子查询
    )

    # 2. 从子查询中选择 rn=1 的记录，即每个分组（subject_id）的最新记录
    #    使用 aliased(Visit, subquery) 来将子查询的结果映射回 Visit 模型
    LatestVisitAlias = aliased(Visit, subquery) # 创建 Visit 模型的一个别名，用于引用子查询的列

    # 3. 构建最终查询，选择 rn=1 的最新访问记录
    #    可以进一步 join Subject 来获取主体信息
    final_query = (
        select(Subject, LatestVisitAlias)
        .join_from(Subject, LatestVisitAlias, Subject.id == LatestVisitAlias.subject_id)
        .where(subquery.c.rn == 1) # 筛选出每个分组中行号为1的记录
        .order_by(Subject.id) # 可选：按主体ID排序结果
    )

    print("\n--- 每个主体的最新访问记录 (使用ROW_NUMBER()) ---")
    results = session.execute(final_query).all()

    for subject, visit in results:
        print(f"主体: {subject.first_name} {subject.last_name}, 最新访问: {visit.date.strftime('%Y-%m-%d')}")

with Session(engine) as session:
    # 构建一个子查询，用于找到每个 subject_id 的最大日期
    latest_date_subquery = (
        select