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
.
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.