在 FastAPI 中使用 SQLAlchemy 的 async 架構時，select().where() 是查詢的標準寫法之一。

📦 範例模型

from sqlalchemy import String, Integer, Column
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String)

🧪 查詢方式（async）

from sqlalchemy import select

async def get_user_by_name(name: str, session: AsyncSession):
    stmt = select(User).where(User.name == name)
    result = await session.execute(stmt)
    return result.scalar_one_or_none()
  • select(User)：選出 User 模型
  • .where(User.name == name)：加條件
  • session.execute(stmt)：執行 SQL
  • .scalar_one_or_none()：取出單一結果（或 None）

🧱 錯誤與調整

  • 記得用 await
  • session 必須是 AsyncSession
  • .first() 也能用，但 .scalar_one_or_none() 更嚴謹

💡 補充

若要查多筆，用 scalars().all() 取出清單：

result = await session.execute(select(User).where(User.name.like("%kat%")))
users = result.scalars().all()