本文 AI 產出,尚未審核
Python 進階主題與實務應用
Context Manager(with 與 __enter__ / __exit__)
簡介
在日常開發中,我們常會需要 「使用完資源後一定要釋放」 的情境,例如開啟檔案、取得資料庫連線、鎖定執行緒或是暫時改變全域設定。若僅靠 try…finally 手寫釋放程式碼,除了程式碼冗長,還容易因為疏忽而遺漏釋放步驟,導致資源泄漏、檔案鎖死或死鎖等問題。
Python 於 2.5 版引入了 Context Manager 機制,讓開發者可以把「取得資源」與「釋放資源」的邏輯封裝在同一個物件裡,並透過 with 陳述式自動呼叫 __enter__、__exit__ 兩個特殊方法。這不僅讓程式碼更簡潔、更具可讀性,也大幅降低錯誤發生的機率。
本篇文章將從 概念、實作方式、常見陷阱、最佳實踐 以及 實務應用場景 逐層說明,幫助你在 Python 專案中熟練運用 Context Manager。
核心概念
1. with 陳述式的工作原理
with 陳述式的語法如下:
with expression as variable:
# 執行區塊
背後的執行流程可以概括為:
- 求值
expression,取得一個支援 Context Manager 介面的物件(必須實作__enter__與__exit__)。 - 呼叫
obj.__enter__(),其回傳值(若有)會指定給as後的變數。 - 執行
with區塊 的程式碼。 - 無論區塊內是否拋出例外,
obj.__exit__(exc_type, exc_val, exc_tb)都會被呼叫。- 若
__exit__回傳True,例外會被「吞掉」;否則例外會繼續向上拋出。
- 若
這個流程等同於手寫的 try…finally,但更具可讀性與可重用性。
2. 實作一個最簡單的 Context Manager
最小的 Context Manager 只需要實作 __enter__ 與 __exit__ 兩個方法:
class SimpleCM:
def __enter__(self):
print(">>> 進入資源")
return self # 可回傳任意物件,供 as 使用
def __exit__(self, exc_type, exc_val, exc_tb):
print("<<< 離開資源")
# 不處理例外,回傳 False 讓例外繼續傳遞
return False
使用方式:
with SimpleCM() as cm:
print("在區塊內執行")
# raise ValueError("測試例外") # 取消註解可觀察 __exit__ 的行為
執行結果:
>>> 進入資源
在區塊內執行
<<< 離開資源
3. contextlib:快速建立 Context Manager
Python 標準庫的 contextlib 提供了兩個便利工具:
| 工具 | 用途 | 範例 |
|---|---|---|
contextmanager |
以 generator 方式寫 __enter__ / __exit__ |
參考範例 4 |
closing |
為只提供 close() 方法的物件包裝成 CM |
with closing(urlopen(...)) as f: |
範例 4:使用 @contextmanager 包裝檔案寫入
from contextlib import contextmanager
@contextmanager
def open_file(path, mode='r', encoding='utf-8'):
f = open(path, mode, encoding=encoding)
try:
yield f # 交給 with 區塊使用
finally:
f.close()
print(f"檔案 {path} 已關閉")
with open_file('demo.txt', 'w') as f:
f.write('Hello, Context Manager!\n')
4. 內建的 Context Manager
Python 已經為許多常見資源提供了內建的 CM,例如:
| 內建 CM | 典型用途 |
|---|---|
open() |
檔案讀寫 |
threading.Lock() |
執行緒同步 |
decimal.localcontext() |
暫時改變 Decimal 計算環境 |
tempfile.TemporaryFile() |
建立臨時檔案 |
sqlite3.connect() |
SQLite 連線管理 |
範例 5:使用 threading.Lock 防止競爭條件
import threading
counter = 0
lock = threading.Lock()
def increment():
global counter
with lock: # 取得 lock,離開時自動釋放
tmp = counter
tmp += 1
counter = tmp
# 多執行緒測試
threads = [threading.Thread(target=increment) for _ in range(1000)]
[t.start() for t in threads]
[t.join() for t in threads]
print(f"最終計數值: {counter}") # 應該是 1000
5. 自訂資源管理:資料庫連線範例
以下示範如何為 SQLite 包裝一個簡易的 Context Manager,讓所有 CRUD 操作自動提交或回滾。
import sqlite3
from contextlib import contextmanager
@contextmanager
def sqlite_conn(db_path):
conn = sqlite3.connect(db_path)
try:
yield conn
conn.commit() # 正常結束時提交
except Exception as e:
conn.rollback() # 發生例外時回滾
print(f"例外: {e},已回滾")
raise
finally:
conn.close()
print("資料庫連線已關閉")
# 使用範例
with sqlite_conn('sample.db') as conn:
cur = conn.cursor()
cur.execute('CREATE TABLE IF NOT EXISTS user(id INTEGER PRIMARY KEY, name TEXT)')
cur.execute('INSERT INTO user(name) VALUES (?)', ('Alice',))
cur.execute('SELECT * FROM user')
rows = cur.fetchall()
print(rows) # [(1, 'Alice')]
常見陷阱與最佳實踐
| 陷阱 | 說明 | 解決方式 |
|---|---|---|
忘記回傳 True 以抑制例外 |
__exit__ 若回傳 True,例外會被吞掉;不小心回傳 True 會隱藏錯誤。 |
只在確定要捕捉例外時才回傳 True,否則回傳 False(或不回傳)。 |
在 __enter__ 內拋出例外,__exit__ 不會執行 |
若取得資源失敗,__exit__ 不會被呼叫,需自行清理。 |
在 __enter__ 中使用 try…except 包住可能失敗的程式,或在 __exit__ 前先檢查資源是否已建立。 |
__exit__ 內再次拋出例外 |
會覆寫原本的例外,導致原始錯誤資訊遺失。 | 只在必要時拋出,或使用 raise 重新拋出原例外。 |
| 同一個 Context Manager 被多次嵌套使用 | 若實作不當,可能導致資源重複釋放或狀態錯亂。 | 在 __enter__ / __exit__ 中使用計數器或 contextvars 追蹤嵌套層級。 |
忘記 yield 前的清理 |
使用 @contextmanager 時,finally 區塊必須放在 yield 之後。 |
確保 yield 前只做「取得資源」的工作,finally 內才釋放。 |
最佳實踐
- 保持
__enter__輕量:只做資源取得與簡單初始化,避免在此階段拋出複雜例外。 - 在
__exit__中統一釋放:即使__enter__失敗,也要確保已取得的部分得到妥善清理。 - 使用
contextlib.ExitStack處理多個資源:當一次需要管理多個 Context Manager 時,ExitStack可動態加入與撤除。 - 文件化返回值:若
__enter__回傳的物件不是self,請在 docstring 明確說明其型別與用途。 - 遵守「一次只做一件事」原則:每個 Context Manager 應聚焦於單一資源或單一行為,讓組合使用更具彈性。
實際應用場景
| 場景 | 為何使用 Context Manager | 範例程式碼 |
|---|---|---|
| 檔案批次處理 | 自動關閉檔案,避免「Too many open files」錯誤。 | with open('log.txt') as f: ... |
| 資料庫交易 | 提交或回滾交易,確保資料一致性。 | 前文的 sqlite_conn 示例。 |
| 網路請求 | requests.Session 需要在使用完畢後關閉連線池。 |
with requests.Session() as s: s.get(url) |
| 測試環境設定 | 暫時改變環境變數或全域設定,測試結束自動還原。 | with patch.dict(os.environ, {'DEBUG': '1'}): ... |
| 多執行緒同步 | Lock、RLock、Semaphore 等鎖定機制。 |
前文的 threading.Lock 示例。 |
| 臨時檔案/目錄 | 使用 tempfile.TemporaryDirectory 產生測試用目錄,結束自動刪除。 |
with tempfile.TemporaryDirectory() as td: ... |
| 複雜資源組合 | ExitStack 同時管理檔案、鎖、資料庫連線等多個資源。 |
with ExitStack() as stack: f = stack.enter_context(open('a.txt')) |
總結
- Context Manager 為 Python 提供了「取得 → 使用 → 釋放」的統一模式,讓資源管理更安全、更易讀。
- 只要實作
__enter__與__exit__,或使用contextlib的裝飾器,就能快速打造自訂的資源管理器。 - 在實務開發中,檔案、資料庫、網路連線、執行緒鎖 等常見資源都適合使用
with陳述式;配合ExitStack更能彈性管理多個資源。 - 注意 例外處理、回傳值 以及 資源的正確釋放,避免常見陷阱。
掌握了 Context Manager 後,你的 Python 程式碼將會變得更「乾淨」與「可靠」,也更符合 PEP 343(即 with 語法的正式規範)所倡導的「可讀性」與「一致性」原則。現在就把這些概念應用到自己的專案中,讓程式碼品質邁向新高度吧!