如何正确使用 SQLModel 和 FastAPI 的异步

Question

下面的代码是使用 FastAPI 和 SQLModel 的简单 API 的示例代码。我在 crud.py 文件中有 crud 函数。我试图在我的函数中实现异步，但出现此错误：

town = db.execute(select(Town).where(Town.id == town_id)).first()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'coroutine' object has no attribute 'first'

这是功能：

async def get_town(db: Session, town_id: int) -> Optional[Town]: 
     """ 
     The get_town function returns a town object from the database. 
  
     :param db: Session: Pass the database session into the function 
     :param town_id: int: Filter the town by id 
     :return: An object of the town class 
     """ 
     town = await db.execute(select(Town).where(Town.id == town_id)) 
     town_result = await town.fetchone() 
     print(town_result) 
     return town_result

虽然当我尝试获取所有城镇时异步实现有效：

async def get_towns(db: Session, skip: int = 0, limit: int = 10) -> List[Town]: 
     """ 
     The get_towns function returns a list of towns from the database. 
  
     :param db: Session: Pass the database session to the function 
     :param skip: int: Skip a number of rows in the database 
     :param limit: int: Limit the number of results returned 
     :return: A list of town objects 
     """ 
     query = select(Town).offset(skip).limit(limit) 
     result = await db.execute(query) 
     return result.scalars().all()

有办法解决这个错误吗？我对使用 SQLModel 还很陌生。

Answer 1

这里需要等待才能打电话

first()

:

town = (await db.execute(select(Town).where(Town.id == town_id))).first()

这里你不需要等待，因为

fetchone()

不是协程：

town = await db.execute(select(Town).where(Town.id == town_id)) 
town_result = town.fetchone()