Search code examples
pythonasync-awaitsqlalchemyfastapiattributeerror

Why doesn't SQLAlchemy's fetchall() method return attributes associated with table fields?


I'm trying to retrieve data from a PostgreSQL table using SQLAlchemy and FastAPI (async). This is my model:

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

Below is a snippet from the user service:

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]
    }

When I make a request, I get an error: AttributeError: user_agent. As far as I understand, the fetchall() method should return a list of Row objects containing attributes associated with the table fields (in my case, user_agent, login_dt etc.). What do I do wrong? I appreciate your replies very much.

Versions I'm using: SQLAlchemy==2.0.16, FastAPI==0.97.0, asyncpg==0.27.0.


Solution

  • The problem is that .fetchall() is returning a list of single-element tuples, where each element is an instance of LoginHistory:

    [
        (<__main__.LoginHistory object at 0x7f43764f7590>,),
        (<__main__.LoginHistory object at 0x7f43764f7610>,)
    ]
    

    so record.user_Agent for record in login_history will fail because the tuples do not have a useragent attribute.

    Calling .scalars() rather than .fetchall() will fix the problem, because the purpose of .scalars() is precisely to eliminate the wrapper tuples from the result. Or we can simply do

    login_history = await session.scalars(query)
    

    and remove the redundant history variable.