SQLAlchemy fetchall()无法返回表字段属性？异步场景求助

我正尝试使用SQLAlchemy和FastAPI（异步）从PostgreSQL表中获取数据。模型定义如下：

class LoginHistory(Base):
    __tablename__ = 'login_history'

    id = Column(
        UUID(as_uuid=True), primary_key=True,
        default=uuid.uuid4, unique=True, nullable=False
    )
    user_id = Column(UUID(as_uuid=True), ForeignKey('user.id'))
    user = relationship('User', back_populates='login_history')
    user_agent = Column(String(255))
    login_dt = Column(DateTime)

    def __init__(self, user_id: UUID, user_agent: str, login_dt: datetime) -> None:
        self.user_id = user_id
        self.login_dt = login_dt
        self.user_agent = user_agent

用户服务代码片段：

from typing import Annotated

from fastapi import Header, Request
from sqlalchemy import insert, select
from sqlalchemy.ext.asyncio import AsyncSession

from auth.src.models.entity import LoginHistory

async def get_login_history(
    authorization: Annotated[str, Header()],
    db: AsyncSession
) -> dict:
    result = await token_logic.get_token_authorization(authorization)
    if result.get('error'):
        return result
    access_token = result.get('token')
    user_id = await token_logic.get_user_id_by_token(access_token)
    query = select(LoginHistory).where(LoginHistory.user_id == user_id)
    history = await db.execute(query)
    login_history = history.fetchall()
    return {'success': [{
            'user_agent': record.user_agent,
            'login_dt': record.login_dt.isoformat()
    } for record in login_history]
    }

请求时出现错误：AttributeError: user_agent。使用版本：SQLAlchemy==2.0.16，FastAPI==0.97.0，asyncpg==0.27.0。

问题原因 #

在SQLAlchemy 2.0中，db.execute(query)返回的Result对象，当执行select(LoginHistory)这类查询整个实体的语句时，fetchall()会返回包含实体实例的Row对象列表，每个Row是元组结构（比如(LoginHistory(id=..., user_agent=...),)），而非直接返回LoginHistory实例。直接访问record.user_agent会报错，因为Row对象本身没有该属性。

两种解决方法 #

1. 使用scalars().all()获取实体实例列表
   将history.fetchall()替换为history.scalars().all()，scalars()会提取Result中的实体实例，all()返回完整列表：

   async def get_login_history(
       authorization: Annotated[str, Header()],
       db: AsyncSession
   ) -> dict:
       # 其他代码不变
       query = select(LoginHistory).where(LoginHistory.user_id == user_id)
       history = await db.execute(query)
       login_history = history.scalars().all()  # 修改此处
       return {'success': [{
               'user_agent': record.user_agent,
               'login_dt': record.login_dt.isoformat()
       } for record in login_history]
       }
   

2. 遍历Row对象时提取第一个元素
   若坚持使用fetchall()，需从每个Row元组中取出第一个元素（即LoginHistory实例）再访问属性：

   async def get_login_history(
       authorization: Annotated[str, Header()],
       db: AsyncSession
   ) -> dict:
       # 其他代码不变
       query = select(LoginHistory).where(LoginHistory.user_id == user_id)
       history = await db.execute(query)
       login_history = history.fetchall()
       return {'success': [{
               'user_agent': record[0].user_agent,  # 添加[0]提取实体
               'login_dt': record[0].login_dt.isoformat()
       } for record in login_history]
       }
   

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