Python 的 async / await 語法，是為了解決「I/O 等待時會卡住整個流程」的問題。

🧠 什麼是非同步（Asynchronous）？

非同步的核心目的：不要讓程式傻傻等，改讓它去做別的事。

例如：發送 API 請求時，我不希望主執行緒被卡住，而是讓它能「邊等邊做別的事」。

✅ 基本語法介紹

import asyncio

async def say_hello():
    await asyncio.sleep(1)
    print("Hello")

async def main():
    await say_hello()

asyncio.run(main())
  • async def：定義一個「非同步函數」
  • await：在呼叫另一個 async 函數時使用，表示「等它做完再繼續」

🔁 常見錯誤

  • ❌ 不能在非 async 函數中用 await
  • ❌ 忘記用 asyncio.run() 來執行

🧪 實際應用場景

  • 呼叫外部 API
  • 檔案上傳下載
  • DB 操作（搭配 async ORM）

🧩 async 與資料庫操作：實務重點

當你使用 FastAPI 並搭配 async ORM（例如 SQLAlchemy 2.x + asyncpg），你會發現：

  • async def 不只是處理外部 API，它也用於所有 I/O 動作，例如「資料庫查詢」。

📂 舉例：db.py 中的設定

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = async_sessionmaker(bind=engine, expire_on_commit=False)

每一個需要連資料庫的函式，都必須是 async 的：

async def get_user_by_id(user_id: int):
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User).where(User.id == user_id))
        return result.scalar_one_or_none()

這就是 async 真正進入實務開發的重要點：

  • 所有涉及資料庫操作的地方，都要用 await
  • 你無法混用 sync ORM 與 async ORM（要選邊站）
  • 如果你 db.py 寫錯模式（用 sync engine），整個專案會無法正常 await

🧘‍♀️ 心得

整個系統流程可以一致處理 async 的流程，包括 API、DB、背景任務。