1

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.

Michael_ON
  • 13
  • 4
  • 1
    Have you tried doing things like printing what is in the variables at different stages in the code - that would be my first port of call - see where things aren't adhering to your assumptions. Could login_history for example be None? – JGFMK Aug 27 '23 at 11:27
  • Fetchall can also return None - it's an Optional List – JGFMK Aug 27 '23 at 11:30
  • Did you add the `user_agent` column to your model after the table had been originally created? – snakecharmerb Aug 27 '23 at 11:54
  • @JGFMK, no it can't. Login_history does contain a single row so far. The printed out content of the variable 'history' is ``. When I try to print out 'login-history', it throws the following error: `sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place?` – Michael_ON Aug 27 '23 at 12:25
  • @snakecharmerb, no, the column 'user_agent' had been added to the model before the table was originally created. – Michael_ON Aug 27 '23 at 12:27
  • @JGFMK, I added more rows to the table `login_history`, the result's the same. By the way, the method `first()` works fine. – Michael_ON Aug 27 '23 at 12:34
  • https://stackoverflow.com/questions/74252768/missinggreenlet-greenlet-spawn-has-not-been-called – JGFMK Aug 27 '23 at 13:00
  • @JGFMK, thank you very much for the link, it was helpful. As far as I got it, the problem was in using `relationship()` in my model, so I would need to use `selectinload` in my query. But I chose another option: I changed my query so that only fields that I need are selected: `query = select(LoginHistory.user_agent, LoginHistory.login_dt).where(LoginHistory.user_id == user_id)`. Thus I avoided selection of the one-to-one related field `user`. Everything worked. Thank you again for the hint. – Michael_ON Aug 27 '23 at 16:33

1 Answers1

2

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.

snakecharmerb
  • 47,570
  • 11
  • 100
  • 153