測驗：認識 asyncpg – Python 非同步 PostgreSQL 的首選

共 5 題，點選答案後會立即顯示結果

1. 非同步資料庫操作相比同步操作的主要優勢是什麼？

A. 程式碼更簡單易懂
B. 可以同時執行多個查詢，減少總等待時間
C. 不需要關閉資料庫連線
D. 可以使用 %s 參數語法

2. asyncpg 使用什麼語法來綁定查詢參數？

await conn.fetch(‘SELECT * FROM users WHERE id = ??? AND status = ???’, 1, ‘active’)

A. %s, %s
B. ?, ?
C. $1, $2
D. :1, :2

3. 如果只需要取得查詢結果的單一值（例如 COUNT(*) 的結果），應該使用哪個方法？

A. conn.fetch()
B. conn.fetchrow()
C. conn.execute()
D. conn.fetchval()

4. asyncpg 的 Record 物件支援以下哪種存取方式？

A. row.name（用屬性存取）
B. row['name'] 和 row[0]（用字典 key 或 index）
C. 只能用 row.get('name')
D. 只能轉成字典後才能存取

5. 在什麼情況下應該優先選擇 asyncpg 而非 psycopg2？

A. 使用 Django 框架開發
B. 撰寫簡單的一次性腳本
C. 使用 FastAPI 開發需要高併發的 Web API
D. 需要原生支援 SQLAlchemy ORM

一句話說明

asyncpg 是專為 Python asyncio 設計的 PostgreSQL 驅動程式，比傳統同步方式快很多。

為什麼需要非同步資料庫操作？

先看一個問題情境：

# 同步方式：一個一個等
user = db.get_user(1)      # 等 50ms
orders = db.get_orders(1)   # 再等 50ms
products = db.get_products() # 再等 50ms
# 總共等了 150ms

# 非同步方式：同時等
user, orders, products = await asyncio.gather(
    db.get_user(1),         # 同時
    db.get_orders(1),       # 同時
    db.get_products()       # 同時
)
# 總共只等 50ms（最慢的那個）
Code language: PHP (php)

翻譯：同步就像排隊買東西，一個一個來；非同步就像同時派三個人去買，誰先買到誰先回來。

什麼時候該用非同步？

情境	用同步	用非同步
簡單腳本、一次性任務	V
Web API（FastAPI、Sanic）	V
同時處理多個請求	V
需要高併發	V

asyncpg vs psycopg2：選哪個？

比較項目	psycopg2	asyncpg
模式	同步（blocking）	非同步（async）
效能	基準	快 5 倍左右
高併發	會卡住	輕鬆應對
連線池	需要額外套件	內建
Django/SQLAlchemy	原生支援	需要額外設定
學習曲線	簡單	需要懂 async/await

一句話選擇指南：

用 Django → 用 psycopg2
用 FastAPI → 用 asyncpg
不確定 → 先看你的框架支援什麼

30 秒安裝與第一次連線

安裝

# 使用 uv（推薦）
uv pip install asyncpg

# 或使用 pip
pip install asyncpg
Code language: PHP (php)

最小可運行範例

import asyncio
import asyncpg

async def main():
    # 建立連線
    conn = await asyncpg.connect(
        'postgresql://postgres:password@localhost/mydb'
    )

    # 執行查詢
    row = await conn.fetchrow('SELECT 1 as num, $1 as name', 'Alice')
    print(row)  # <Record num=1 name='Alice'>

    # 關閉連線
    await conn.close()

# 執行
asyncio.run(main())
Code language: PHP (php)

這段代碼做了什麼：

asyncpg.connect() – 連線到 PostgreSQL
conn.fetchrow() – 執行查詢，取得一筆資料
conn.close() – 用完要關閉連線

核心概念翻譯

你會看到	意思
`await asyncpg.connect(...)`	建立一個資料庫連線
`await conn.fetch(...)`	查詢並取得多筆資料（List）
`await conn.fetchrow(...)`	查詢並取得一筆資料
`await conn.fetchval(...)`	查詢並取得單一值
`await conn.execute(...)`	執行 SQL（不需要回傳資料時用）
`$1, $2, $3`	參數佔位符（不是 %s）

AI 最常這樣用

用法 1：查詢多筆資料

async def get_all_users(conn):
    # fetch 回傳 List[Record]
    rows = await conn.fetch('SELECT id, name FROM users')

    for row in rows:
        print(row['id'], row['name'])  # 用字典方式存取
Code language: PHP (php)

翻譯：fetch 會回傳一個列表，裡面每個元素都是一筆資料。

用法 2：查詢單筆資料

async def get_user(conn, user_id: int):
    # fetchrow 回傳單一 Record（或 None）
    row = await conn.fetchrow(
        'SELECT * FROM users WHERE id = $1',
        user_id
    )

    if row:
        return dict(row)  # 轉成普通字典
    return None
Code language: PHP (php)

翻譯：fetchrow 只拿一筆，找不到會回傳 None。

用法 3：只要一個值

async def count_users(conn):
    # fetchval 只回傳第一個欄位的值
    count = await conn.fetchval('SELECT COUNT(*) FROM users')
    return count  # 直接是數字，不是 Record
Code language: PHP (php)

翻譯：fetchval 直接給你值，不用再從 Record 裡取。

用法 4：執行不需要回傳的操作

async def create_user(conn, name: str, email: str):
    # execute 用於 INSERT/UPDATE/DELETE
    await conn.execute(
        'INSERT INTO users (name, email) VALUES ($1, $2)',
        name, email
    )
Code language: PHP (php)

翻譯：新增、修改、刪除用 execute，不會回傳資料。

Record 物件怎麼用

asyncpg 查詢回來的不是普通字典，是 Record 物件：

row = await conn.fetchrow('SELECT id, name, email FROM users WHERE id = $1', 1)

# 三種存取方式都可以
print(row['name'])     # 用 key 存取（像字典）
print(row[1])          # 用 index 存取（像 tuple）
print(row.get('name')) # 用 get 存取（找不到回傳 None）

# 轉成普通字典
user_dict = dict(row)
print(user_dict)  # {'id': 1, 'name': 'Alice', 'email': '[email protected]'}
Code language: PHP (php)

Record vs 字典

操作	Record	dict
`row['name']`	V	V
`row[0]`	V	X
`row.get('name')`	V	V
轉成字典	`dict(row)`	本身就是

Vibe Coder 提示：如果 AI 寫的代碼用 row.name 會報錯，Record 不支援這種語法，要用 row['name']。

參數綁定：$1, $2, $3

asyncpg 使用 $n 語法綁定參數，不是 Python 常見的 %s 或 ?：

# asyncpg 的方式
await conn.fetch('SELECT * FROM users WHERE id = $1 AND status = $2', 1, 'active')
#                                                  ↑         ↑      ↑     ↑
#                                                 $1        $2     對應    對應

# 不是這樣（這會錯）
await conn.fetch('SELECT * FROM users WHERE id = %s', (1,))  # 錯！
await conn.fetch('SELECT * FROM users WHERE id = ?', 1)      # 錯！
Code language: PHP (php)

一句話：$1 對應第一個參數，$2 對應第二個，以此類推。

完整範例：CRUD 操作

import asyncio
import asyncpg

async def main():
    # 連線
    conn = await asyncpg.connect('postgresql://postgres:password@localhost/mydb')

    # Create - 新增
    await conn.execute(
        'INSERT INTO users (name, email) VALUES ($1, $2)',
        'Bob', '[email protected]'
    )

    # Read - 查詢單筆
    user = await conn.fetchrow('SELECT * FROM users WHERE name = $1', 'Bob')
    print(f"找到用戶: {user['name']}")

    # Read - 查詢多筆
    all_users = await conn.fetch('SELECT * FROM users')
    print(f"共有 {len(all_users)} 位用戶")

    # Update - 更新
    await conn.execute(
        'UPDATE users SET email = $1 WHERE name = $2',
        '[email protected]', 'Bob'
    )

    # Delete - 刪除
    await conn.execute('DELETE FROM users WHERE name = $1', 'Bob')

    # 關閉連線
    await conn.close()

asyncio.run(main())
Code language: PHP (php)

Vibe Coder 檢查點

看到 AI 用 asyncpg 時確認：

[ ] 有沒有 await？ – 所有 asyncpg 操作都要加 await
[ ] 連線有沒有關閉？ – 用完要 await conn.close() 或用 context manager
[ ] 參數用 $1, $2？ – 不是 %s 或 ?
[ ] fetchrow 有沒有處理 None？ – 查無資料會回傳 None
[ ] Record 存取方式對嗎？ – 用 row['column'] 不是 row.column

常見錯誤與解法

錯誤 1：忘記 await

# 錯誤
row = conn.fetchrow('SELECT * FROM users')  # 少了 await
print(row)  # 會印出 coroutine 物件，不是資料

# 正確
row = await conn.fetchrow('SELECT * FROM users')
Code language: PHP (php)

錯誤 2：用錯參數語法

# 錯誤
await conn.fetch('SELECT * FROM users WHERE id = %s', 1)  # psycopg2 語法

# 正確
await conn.fetch('SELECT * FROM users WHERE id = $1', 1)  # asyncpg 語法
Code language: PHP (php)

錯誤 3：在同步函式裡呼叫 async

# 錯誤
def get_user():  # 不是 async
    return await conn.fetchrow(...)  # 會報錯

# 正確
async def get_user():  # 加上 async
    return await conn.fetchrow(...)
Code language: PHP (php)

延伸：知道就好

這些進階功能之後會教，現在先知道存在就好：

連線池（Connection Pool）：管理多個連線，適合 Web 應用
交易（Transaction）：多個操作要同時成功或失敗
Prepared Statements：重複查詢時提升效能
COPY 操作：大量資料匯入匯出

下一步

學完這篇，你已經會：

理解同步 vs 非同步的差異
安裝 asyncpg 並建立連線
使用 fetch / fetchrow / fetchval / execute
正確使用 $1, $2 參數綁定
處理 Record 物件

下一篇我們會學習「連線池」- 在 Web 應用中正確管理資料庫連線的方式。

參考資源

進階測驗：認識 asyncpg – Python 非同步 PostgreSQL 的首選

測驗目標：驗證你是否能在實際情境中應用所學。
共 5 題，包含情境題與錯誤診斷題。

1. 你正在開發一個用戶儀表板 API，需要同時取得用戶資料、訂單列表和產品清單。如何用 asyncpg 最有效率地完成這個需求？情境題

A. 依序呼叫三個 await conn.fetch()
B. 使用 asyncio.gather() 同時執行三個查詢
C. 使用一個 JOIN 將三個表合併查詢
D. 建立三個獨立的資料庫連線分別查詢

2. 小明寫了以下程式碼，但執行後印出的是一個奇怪的物件而不是資料。問題出在哪？錯誤診斷

async def get_user(conn, user_id):
    row = conn.fetchrow(‘SELECT * FROM users WHERE id = $1’, user_id)
    print(row)
    return row

A. 參數語法錯誤，應該用 %s 而不是 $1
B. fetchrow 應該改用 fetch
C. 忘記在 conn.fetchrow() 前加 await
D. user_id 參數位置錯誤

3. 你需要實作一個函式來計算資料庫中的用戶總數，並直接回傳數字。以下哪個實作最合適？情境題

A. rows = await conn.fetch('SELECT COUNT(*) FROM users'); return len(rows)
B. row = await conn.fetchrow('SELECT COUNT(*) FROM users'); return row[0]
C. await conn.execute('SELECT COUNT(*) FROM users')
D. count = await conn.fetchval('SELECT COUNT(*) FROM users'); return count

4. 小華的程式碼在執行時拋出錯誤。請判斷問題所在：錯誤診斷

row = await conn.fetchrow(‘SELECT id, name, email FROM users WHERE id = $1’, 1)
print(f”User name: {row.name}”)

A. 查詢語法錯誤
B. Record 不支援 row.name 屬性存取，應改用 row['name']
C. 缺少 await 關鍵字
D. 應該先用 dict(row) 轉換才能存取

5. 你在建立一個 FastAPI 應用，需要將新用戶資料寫入資料庫。用戶資料包含 name 和 email。以下哪個寫法正確？情境題

async def create_user(conn, name: str, email: str):
    # 選擇正確的實作方式

A. await conn.fetch('INSERT INTO users (name, email) VALUES (%s, %s)', name, email)
B. await conn.fetchrow('INSERT INTO users (name, email) VALUES ($1, $2)', name, email)
C. await conn.execute('INSERT INTO users (name, email) VALUES ($1, $2)', name, email)
D. await conn.fetchval('INSERT INTO users (name, email) VALUES ($1, $2)', name, email)

【asyncpg 教學】#01 認識 asyncpg：Python 非同步 PostgreSQL 的首選

測驗：認識 asyncpg – Python 非同步 PostgreSQL 的首選

1. 非同步資料庫操作相比同步操作的主要優勢是什麼？

2. asyncpg 使用什麼語法來綁定查詢參數？

3. 如果只需要取得查詢結果的單一值（例如 COUNT(*) 的結果），應該使用哪個方法？

4. asyncpg 的 Record 物件支援以下哪種存取方式？

5. 在什麼情況下應該優先選擇 asyncpg 而非 psycopg2？

一句話說明

為什麼需要非同步資料庫操作？

什麼時候該用非同步？

asyncpg vs psycopg2：選哪個？

30 秒安裝與第一次連線

安裝

最小可運行範例

核心概念翻譯

AI 最常這樣用

用法 1：查詢多筆資料

用法 2：查詢單筆資料

用法 3：只要一個值

用法 4：執行不需要回傳的操作

Record 物件怎麼用

Record vs 字典

參數綁定：$1, $2, $3

完整範例：CRUD 操作

Vibe Coder 檢查點

常見錯誤與解法

錯誤 1：忘記 await

錯誤 2：用錯參數語法

錯誤 3：在同步函式裡呼叫 async

延伸：知道就好

下一步

參考資源

進階測驗：認識 asyncpg – Python 非同步 PostgreSQL 的首選

1. 你正在開發一個用戶儀表板 API，需要同時取得用戶資料、訂單列表和產品清單。如何用 asyncpg 最有效率地完成這個需求？情境題

2. 小明寫了以下程式碼，但執行後印出的是一個奇怪的物件而不是資料。問題出在哪？錯誤診斷

3. 你需要實作一個函式來計算資料庫中的用戶總數，並直接回傳數字。以下哪個實作最合適？情境題

4. 小華的程式碼在執行時拋出錯誤。請判斷問題所在：錯誤診斷

5. 你在建立一個 FastAPI 應用，需要將新用戶資料寫入資料庫。用戶資料包含 name 和 email。以下哪個寫法正確？情境題

發佈留言取消回覆

測驗：認識 asyncpg – Python 非同步 PostgreSQL 的首選

1. 非同步資料庫操作相比同步操作的主要優勢是什麼？

2. asyncpg 使用什麼語法來綁定查詢參數？

3. 如果只需要取得查詢結果的單一值（例如 COUNT(*) 的結果），應該使用哪個方法？

4. asyncpg 的 Record 物件支援以下哪種存取方式？

5. 在什麼情況下應該優先選擇 asyncpg 而非 psycopg2？

一句話說明

為什麼需要非同步資料庫操作？

什麼時候該用非同步？

asyncpg vs psycopg2：選哪個？

30 秒安裝與第一次連線

安裝

最小可運行範例

核心概念翻譯

AI 最常這樣用

用法 1：查詢多筆資料

用法 2：查詢單筆資料

用法 3：只要一個值

用法 4：執行不需要回傳的操作

Record 物件怎麼用

Record vs 字典

參數綁定：$1, $2, $3

完整範例：CRUD 操作

Vibe Coder 檢查點

常見錯誤與解法

錯誤 1：忘記 await

錯誤 2：用錯參數語法

錯誤 3：在同步函式裡呼叫 async

延伸：知道就好

下一步

參考資源

進階測驗：認識 asyncpg – Python 非同步 PostgreSQL 的首選

1. 你正在開發一個用戶儀表板 API，需要同時取得用戶資料、訂單列表和產品清單。如何用 asyncpg 最有效率地完成這個需求？ 情境題

2. 小明寫了以下程式碼，但執行後印出的是一個奇怪的物件而不是資料。問題出在哪？ 錯誤診斷

3. 你需要實作一個函式來計算資料庫中的用戶總數，並直接回傳數字。以下哪個實作最合適？ 情境題

4. 小華的程式碼在執行時拋出錯誤。請判斷問題所在： 錯誤診斷

5. 你在建立一個 FastAPI 應用，需要將新用戶資料寫入資料庫。用戶資料包含 name 和 email。以下哪個寫法正確？ 情境題

相關文章

asyncpg 進階技巧快速看懂：Vibe Coder 必知

【asyncpg 教學】#03 CRUD 操作與參數化查詢

【asyncpg 教學】#02 連線管理：Connection Pool 實戰

發佈留言取消回覆

1. 你正在開發一個用戶儀表板 API，需要同時取得用戶資料、訂單列表和產品清單。如何用 asyncpg 最有效率地完成這個需求？情境題

2. 小明寫了以下程式碼，但執行後印出的是一個奇怪的物件而不是資料。問題出在哪？錯誤診斷

3. 你需要實作一個函式來計算資料庫中的用戶總數，並直接回傳數字。以下哪個實作最合適？情境題

4. 小華的程式碼在執行時拋出錯誤。請判斷問題所在：錯誤診斷

5. 你在建立一個 FastAPI 應用，需要將新用戶資料寫入資料庫。用戶資料包含 name 和 email。以下哪個寫法正確？情境題