如何在FastAPI与SQLModel中正确使用异步？解决AttributeError问题

错误根源 #

1. 未等待异步方法执行：异步模式下db.execute()是协程函数，必须用await等待执行完成。直接链式调用.first()会操作协程对象而非实际查询结果，因此触发'coroutine' object has no attribute 'first'错误。
2. 结果处理逻辑不当：异步查询返回的AsyncResult对象，直接用fetchone()得到的是数据库行对象，而非映射的Town模型实例，需要通过特定方法提取模型数据。
3. Session类型不匹配：异步CRUD函数应接收AsyncSession而非同步Session，否则会引发异步兼容性问题。

正确实现方式 #

方式一：通过scalars().first()获取单个模型实例

from sqlmodel import select, AsyncSession
from typing import Optional
from models import Town  # 替换为你的模型文件路径

async def get_town(db: AsyncSession, town_id: int) -> Optional[Town]: 
    """根据ID获取单个城镇"""
    result = await db.execute(select(Town).where(Town.id == town_id))
    # scalars()提取映射的模型实例，first()返回第一个结果（无结果则为None）
    return result.scalars().first()

方式二：用db.scalar()简化代码

db.scalar()是异步方法，可直接返回查询的单个标量结果（即Town实例），无需手动处理Result对象：

async def get_town(db: AsyncSession, town_id: int) -> Optional[Town]: 
    return await db.scalar(select(Town).where(Town.id == town_id))

配套异步Session注入（FastAPI路由层） #

确保FastAPI路由中注入的是异步Session，示例：

from fastapi import FastAPI, Depends
from sqlmodel import create_async_engine, AsyncSession
from crud import get_town

# 创建异步数据库引擎（以SQLite为例）
async_engine = create_async_engine("sqlite+aiosqlite:///./towns.db", echo=True)

# 依赖函数：提供异步Session
async def get_db() -> AsyncSession:
    async with AsyncSession(async_engine) as session:
        yield session

app = FastAPI()

@app.get("/town/{town_id}")
async def read_town(town_id: int, db: AsyncSession = Depends(get_db)):
    return await get_town(db, town_id)

为什么get_towns能正常运行？ #

get_towns中result.scalars().all()是符合异步Result对象API的正确写法：scalars()提取所有映射的Town实例，all()返回实例列表，因此可以正常工作。

内容的提问来源于stack exchange，提问作者Ifeanyi Nneji