我正在try 使用SQLAlchemy
和FastAPI
(async)从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
.据我所知,fetchall()
方法应该返回一个包含Row
个对象的列表,其中包含与表字段相关联的属性(在我的例子中,是user_agent
、login_dt
等).我做错了什么?非常感谢您的回复.
我正在使用的版本:SQLAlchemy==2.0.16
、FastAPI==0.97.0
、asyncpg==0.27.0
.