測驗：asyncpg 進階技巧快速看懂

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

1. 在 asyncpg 中，`async with conn.transaction()` 的主要作用是什麼？

A. 加速查詢執行速度
B. 確保區塊內的操作全部成功或全部失敗
C. 自動建立資料庫連線池
D. 預先編譯 SQL 語句

2. 當需要匯入大量資料（例如 10000 筆）到資料庫時，最有效率的做法是？

A. 使用迴圈一筆一筆執行 INSERT
B. 使用 prepare() 預編譯後再執行
C. 使用 copy_records_to_table()
D. 使用多執行緒同時 INSERT

3. 以下程式碼中，`prepare()` 的主要用途是什麼？

stmt = await conn.prepare(“SELECT * FROM users WHERE id = $1”)
user = await stmt.fetchrow(123)

A. 將查詢結果快取在記憶體中
B. 預編譯 SQL 語句，讓之後只需傳參數即可執行
C. 建立新的資料庫連線
D. 驗證 SQL 語法是否正確

4. 在 asyncpg 中，`InterfaceError` 代表什麼類型的錯誤？

A. SQL 語法錯誤
B. 違反唯一性約束（主鍵重複）
C. 連線層面的錯誤（連不上、連線斷了）
D. 違反外鍵約束

5. 在重試策略中使用「指數退避」的目的是什麼？

await asyncio.sleep(2 ** attempt) # 等待時間：1秒、2秒、4秒…

A. 讓程式執行更快
B. 避免連續快速重試而打爆資料庫
C. 確保每次重試都使用不同的連線
D. 讓使用者有時間手動修復問題

一句話說明

交易確保操作全做或全不做，批次匯入加速大量資料處理。

為什麼 Vibe Coder 要懂這些？

當 AI 幫你寫資料庫操作時，它經常會用到交易（transaction）、批次操作、prepared statements 這些進階功能。你需要看懂這些程式碼在做什麼，才能確保資料安全、效能良好。

你會遇到的情況	進階技巧能幫你
轉帳要同時扣款和入帳	用交易確保兩邊同步
要匯入幾萬筆資料	用 COPY 命令加速
同一個查詢跑很多次	用 prepared statement 省效能
資料庫連線斷了	用重試策略自動恢復

交易管理：async with connection.transaction()

最小範例

async with conn.transaction():
    await conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    await conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
Code language: JavaScript (javascript)

逐行翻譯

async with conn.transaction():           # 開始交易，區塊內的操作會一起成功或一起失敗
    await conn.execute("UPDATE ...")     # 第一個操作：扣款
    await conn.execute("UPDATE ...")     # 第二個操作：入帳
# 離開區塊時自動提交，如果中間出錯會自動回滾
Code language: PHP (php)

翻譯：這就像「全有或全無」的保證。如果扣款成功但入帳失敗，扣款也會被取消。

AI 常見寫法

寫法 1：基本交易

async def transfer_money(conn, from_id, to_id, amount):
    async with conn.transaction():
        # 檢查餘額
        balance = await conn.fetchval(
            "SELECT balance FROM accounts WHERE id = $1", from_id
        )
        if balance < amount:
            raise ValueError("餘額不足")

        # 執行轉帳
        await conn.execute(
            "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
            amount, from_id
        )
        await conn.execute(
            "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
            amount, to_id
        )
Code language: PHP (php)

翻譯：先檢查餘額，再執行轉帳。整個過程在交易中，任何一步失敗都會回滾。

寫法 2：使用 Savepoint（部分回滾）

async with conn.transaction():
    await conn.execute("INSERT INTO orders ...")  # 一定要成功

    try:
        async with conn.transaction():  # 巢狀交易 = savepoint
            await conn.execute("INSERT INTO notifications ...")
    except Exception:
        pass  # 通知失敗沒關係，訂單還是會成立
Code language: PHP (php)

翻譯：巢狀的 async with conn.transaction() 會建立 savepoint。內層失敗只會回滾內層，外層繼續執行。

寫法 3：明確控制提交

tr = conn.transaction()
await tr.start()
try:
    await conn.execute("INSERT INTO ...")
    await tr.commit()  # 明確提交
except Exception:
    await tr.rollback()  # 明確回滾
Code language: PHP (php)

翻譯：不用 async with，手動控制交易的開始、提交、回滾。較少見，但有時需要更細緻的控制。

批次匯入：copy_records_to_table()

最小範例

records = [(1, "Alice"), (2, "Bob"), (3, "Charlie")]
await conn.copy_records_to_table("users", records=records)
Code language: JavaScript (javascript)

逐行翻譯

records = [(1, "Alice"), (2, "Bob"), (3, "Charlie")]  # 準備資料，每個 tuple 是一列
await conn.copy_records_to_table(                      # 使用 PostgreSQL COPY 命令
    "users",                                           # 目標資料表
    records=records                                    # 要匯入的資料
)
Code language: PHP (php)

翻譯：這比一筆一筆 INSERT 快非常多（可能快 10-100 倍），適合大量資料匯入。

AI 常見寫法

寫法 1：指定欄位

await conn.copy_records_to_table(
    "users",
    records=records,
    columns=["id", "name"]  # 明確指定欄位順序
)
Code language: PHP (php)

翻譯：當資料表有很多欄位，但你只想匯入部分欄位時使用。

寫法 2：搭配交易

async with conn.transaction():
    await conn.execute("DELETE FROM temp_data")
    await conn.copy_records_to_table("temp_data", records=big_data)
Code language: JavaScript (javascript)

翻譯：先清空再匯入，整個過程在交易中。如果匯入失敗，清空也會被取消。

寫法 3：從檔案匯入

async with open("data.csv", "rb") as f:
    await conn.copy_to_table(
        "users",
        source=f,
        format="csv",
        header=True
    )
Code language: JavaScript (javascript)

翻譯：直接從 CSV 檔案匯入資料表，比先讀檔再處理更有效率。

預編譯查詢：prepare()

最小範例

stmt = await conn.prepare("SELECT * FROM users WHERE id = $1")
user = await stmt.fetchrow(123)
Code language: JavaScript (javascript)

逐行翻譯

stmt = await conn.prepare("SELECT * FROM users WHERE id = $1")  # 預編譯 SQL，資料庫會快取執行計畫
user = await stmt.fetchrow(123)                                   # 用預編譯的語句查詢，只需傳參數
Code language: PHP (php)

翻譯：把 SQL 語句先「編譯」好存起來，之後只需要傳參數。適合同一個查詢要跑很多次的情況。

AI 常見寫法

寫法 1：批次查詢

stmt = await conn.prepare("SELECT * FROM products WHERE id = $1")

for product_id in product_ids:
    product = await stmt.fetchrow(product_id)
    # 處理每個產品
Code language: PHP (php)

翻譯：同一個查詢跑 1000 次，用 prepared statement 比每次都寫完整 SQL 更快。

寫法 2：查看執行計畫

stmt = await conn.prepare("SELECT * FROM users WHERE email = $1")
print(stmt.get_attributes())  # 查看回傳欄位資訊
Code language: PHP (php)

翻譯：可以在不執行查詢的情況下，先知道會回傳哪些欄位。

錯誤處理

常見錯誤類型

import asyncpg

try:
    await conn.execute("SELECT * FROM nonexistent")
except asyncpg.PostgresError as e:
    # 所有 PostgreSQL 錯誤的基類
    print(f"資料庫錯誤：{e}")
except asyncpg.InterfaceError as e:
    # 連線相關錯誤（連線斷了、Pool 關閉了等）
    print(f"連線錯誤：{e}")
Code language: PHP (php)

核心錯誤翻譯

錯誤類型	意思
`PostgresError`	資料庫層面的錯誤（SQL 錯誤、違反約束等）
`InterfaceError`	連線層面的錯誤（連不上、連線斷了）
`UniqueViolationError`	違反唯一性約束（主鍵或 unique 欄位重複）
`ForeignKeyViolationError`	違反外鍵約束
`ConnectionDoesNotExistError`	連線已經關閉

AI 常見錯誤處理模式

模式 1：捕捉特定錯誤

try:
    await conn.execute(
        "INSERT INTO users (email) VALUES ($1)", email
    )
except asyncpg.UniqueViolationError:
    print("這個 email 已經被註冊了")
except asyncpg.PostgresError as e:
    print(f"其他資料庫錯誤：{e}")
Code language: PHP (php)

翻譯：先捕捉最具體的錯誤（唯一性違反），再用通用類型捕捉其他錯誤。

模式 2：交易中的錯誤處理

try:
    async with conn.transaction():
        await conn.execute("INSERT INTO ...")
        await conn.execute("UPDATE ...")
except asyncpg.PostgresError:
    # 不需要手動 rollback，async with 會自動處理
    print("交易失敗，已自動回滾")
Code language: PHP (php)

翻譯：在 async with conn.transaction() 中發生錯誤，會自動回滾，不需要手動處理。

連線中斷與重試策略

最小範例

import asyncio

async def execute_with_retry(pool, query, *args, max_retries=3):
    for attempt in range(max_retries):
        try:
            async with pool.acquire() as conn:
                return await conn.execute(query, *args)
        except asyncpg.InterfaceError:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # 指數退避：1秒、2秒、4秒
Code language: PHP (php)

逐行翻譯

async def execute_with_retry(pool, query, *args, max_retries=3):
    for attempt in range(max_retries):           # 最多重試 3 次
        try:
            async with pool.acquire() as conn:   # 從連線池取得連線
                return await conn.execute(query, *args)  # 執行查詢
        except asyncpg.InterfaceError:           # 連線錯誤（斷線等）
            if attempt == max_retries - 1:       # 最後一次嘗試
                raise                            # 還是失敗就拋出錯誤
            await asyncio.sleep(2 ** attempt)    # 等一下再重試
Code language: PHP (php)

翻譯：連線斷了就重試，每次等待時間加倍（1秒、2秒、4秒），避免一直打資料庫。

AI 常見重試模式

模式 1：使用裝飾器

from functools import wraps

def with_retry(max_retries=3):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                except asyncpg.InterfaceError:
                    if attempt == max_retries - 1:
                        raise
                    await asyncio.sleep(2 ** attempt)
        return wrapper
    return decorator

@with_retry(max_retries=3)
async def get_user(pool, user_id):
    async with pool.acquire() as conn:
        return await conn.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
Code language: JavaScript (javascript)

翻譯：把重試邏輯包成裝飾器，任何函式加上 @with_retry() 就有自動重試功能。

模式 2：連線池健康檢查

pool = await asyncpg.create_pool(
    dsn,
    min_size=5,
    max_size=20,
    command_timeout=60,           # 單一指令超時 60 秒
    max_inactive_connection_lifetime=300  # 閒置連線 5 分鐘後關閉
)
Code language: PHP (php)

翻譯：設定連線池參數，自動管理連線健康狀態，減少連線問題。

Vibe Coder 檢查點

看到 AI 用這些進階功能時確認：

[ ] 用了 transaction() 嗎？需要「全有或全無」的操作應該用交易包起來
[ ] 大量資料用 copyrecordsto_table() 還是一筆一筆 INSERT？前者快很多
[ ] 同一個查詢跑很多次有用 prepare() 嗎？
[ ] 有處理 InterfaceError 連線錯誤嗎？
[ ] 有重試機制嗎？用了指數退避避免打爆資料庫嗎？

必看懂

這些進階功能會一直出現，要熟：

功能	記法
`async with conn.transaction()`	「交易，全做或全不做」
`copy_records_to_table()`	「批次匯入，很快」
`await conn.prepare()`	「預編譯，重複查詢用這個」
`asyncpg.InterfaceError`	「連線壞了」
`asyncpg.PostgresError`	「SQL 或資料庫錯誤」

知道就好

這些進階功能遇到再查：

Cursor：處理超大結果集，一次讀一部分
LISTEN/NOTIFY：資料庫事件通知，即時推送
copyfromquery()：把查詢結果匯出到檔案
Connection.addloglistener()：監聽資料庫日誌
Pool.setconnectargs()：動態更新連線參數

效能對照

讓你知道這些技巧有多重要：

操作	一般做法	進階做法	速度差異
插入 10000 筆	一筆一筆 INSERT	copy_records_to_table	約快 50-100 倍
同查詢跑 1000 次	每次完整 SQL	prepare()	約快 20-30%
轉帳操作	兩個獨立 UPDATE	transaction()	保證資料一致性

小結

asyncpg 進階技巧對 Vibe Coder 來說就是：

交易：用 async with conn.transaction() 確保多個操作一起成功或失敗
批次匯入：用 copyrecordsto_table() 快速匯入大量資料
預編譯：用 prepare() 加速重複執行的查詢
錯誤處理：分清 PostgresError（SQL 錯誤）和 InterfaceError（連線錯誤）
重試策略：連線斷了要重試，用指數退避避免打爆資料庫

記住這五個重點，看到 AI 寫的進階資料庫操作就不會慌了！

進階測驗：asyncpg 進階技巧快速看懂

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

1. 你正在開發一個銀行轉帳功能，需要從帳戶 A 扣款並存入帳戶 B。AI 幫你產生了以下程式碼，你應該如何改進？情境題

async def transfer(conn, from_id, to_id, amount):
    await conn.execute(
        “UPDATE accounts SET balance = balance – $1 WHERE id = $2”,
        amount, from_id
    )
    await conn.execute(
        “UPDATE accounts SET balance = balance + $1 WHERE id = $2”,
        amount, to_id
    )

A. 加上 try/except 捕捉錯誤即可
B. 用 async with conn.transaction() 包住兩個操作
C. 用 prepare() 預編譯 SQL 語句
D. 改用 copy_records_to_table() 批次處理

2. 小明寫了一個程式要匯入 50000 筆使用者資料，執行後非常慢。以下是他的程式碼，問題出在哪裡？錯誤診斷

async def import_users(conn, users):
    for user in users:  # users 有 50000 筆
        await conn.execute(
            “INSERT INTO users (name, email) VALUES ($1, $2)”,
            user[‘name’], user[’email’]
        )

A. 缺少 async with conn.transaction()
B. SQL 語法錯誤導致效能問題
C. 應該使用 copy_records_to_table() 批次匯入
D. 需要先用 prepare() 預編譯

3. 你在開發訂單系統，建立訂單時也要發送通知。但通知失敗不應該影響訂單的建立。以下哪種寫法最適合？情境題

A. 不使用交易，讓兩個操作獨立執行
B. 把通知操作放在 try/except 中但不在交易內
C. 用一個交易包住所有操作，失敗就全部回滾
D. 使用巢狀交易（Savepoint），通知失敗只回滾內層

4. 程式執行時出現以下錯誤，最可能的原因和解決方法是什麼？錯誤診斷

asyncpg.exceptions.UniqueViolationError: duplicate key value violates unique constraint “users_email_key”

A. 資料庫連線斷開，需要重新連線
B. 嘗試插入的 email 已經存在，需要先檢查或使用 ON CONFLICT
C. SQL 語法錯誤，需要修正查詢語句
D. 外鍵約束失敗，需要先插入關聯資料

5. 你的應用程式在資料庫重啟後頻繁出現 `InterfaceError`，且連線失敗後不會自動恢復。你應該如何改進？情境題

async def get_user(pool, user_id):
    async with pool.acquire() as conn:
        return await conn.fetchrow(“SELECT * FROM users WHERE id = $1”, user_id)

A. 改用 prepare() 預編譯查詢
B. 在外層加上 async with conn.transaction()
C. 加入重試機制，捕捉 InterfaceError 並使用指數退避
D. 使用 copy_records_to_table() 改善效能

測驗：asyncpg 進階技巧快速看懂

1. 在 asyncpg 中，async with conn.transaction() 的主要作用是什麼？

2. 當需要匯入大量資料（例如 10000 筆）到資料庫時，最有效率的做法是？

3. 以下程式碼中，prepare() 的主要用途是什麼？

4. 在 asyncpg 中，InterfaceError 代表什麼類型的錯誤？

5. 在重試策略中使用「指數退避」的目的是什麼？

一句話說明

為什麼 Vibe Coder 要懂這些？

交易管理：async with connection.transaction()

最小範例

逐行翻譯

AI 常見寫法

寫法 1：基本交易

寫法 2：使用 Savepoint（部分回滾）

寫法 3：明確控制提交

批次匯入：copy_records_to_table()

最小範例

逐行翻譯

AI 常見寫法

寫法 1：指定欄位

寫法 2：搭配交易

寫法 3：從檔案匯入

預編譯查詢：prepare()

最小範例

逐行翻譯

AI 常見寫法

寫法 1：批次查詢

寫法 2：查看執行計畫

錯誤處理

常見錯誤類型

核心錯誤翻譯

AI 常見錯誤處理模式

模式 1：捕捉特定錯誤

模式 2：交易中的錯誤處理

連線中斷與重試策略

最小範例

逐行翻譯

AI 常見重試模式

模式 1：使用裝飾器

模式 2：連線池健康檢查

Vibe Coder 檢查點

必看懂

知道就好

效能對照

小結

進階測驗：asyncpg 進階技巧快速看懂

1. 你正在開發一個銀行轉帳功能，需要從帳戶 A 扣款並存入帳戶 B。AI 幫你產生了以下程式碼，你應該如何改進？ 情境題

2. 小明寫了一個程式要匯入 50000 筆使用者資料，執行後非常慢。以下是他的程式碼，問題出在哪裡？ 錯誤診斷

3. 你在開發訂單系統，建立訂單時也要發送通知。但通知失敗不應該影響訂單的建立。以下哪種寫法最適合？ 情境題

4. 程式執行時出現以下錯誤，最可能的原因和解決方法是什麼？ 錯誤診斷

5. 你的應用程式在資料庫重啟後頻繁出現 InterfaceError，且連線失敗後不會自動恢復。你應該如何改進？ 情境題

相關文章

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

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

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

發佈留言取消回覆

1. 在 asyncpg 中，`async with conn.transaction()` 的主要作用是什麼？

3. 以下程式碼中，`prepare()` 的主要用途是什麼？

4. 在 asyncpg 中，`InterfaceError` 代表什麼類型的錯誤？

1. 你正在開發一個銀行轉帳功能，需要從帳戶 A 扣款並存入帳戶 B。AI 幫你產生了以下程式碼，你應該如何改進？情境題

2. 小明寫了一個程式要匯入 50000 筆使用者資料，執行後非常慢。以下是他的程式碼，問題出在哪裡？錯誤診斷

3. 你在開發訂單系統，建立訂單時也要發送通知。但通知失敗不應該影響訂單的建立。以下哪種寫法最適合？情境題

4. 程式執行時出現以下錯誤，最可能的原因和解決方法是什麼？錯誤診斷

5. 你的應用程式在資料庫重啟後頻繁出現 `InterfaceError`，且連線失敗後不會自動恢復。你應該如何改進？情境題