檔案 IO 與 with 語句#
讀寫檔案是程式最常見的需求之一。Python 用 open() 開啟檔案,但開啟後 一定要關閉 ——否則會佔用作業系統資源,緩衝區的資料也可能還沒寫進磁碟。with 語句是 Python 的語法糖,能保證檔案在離開區塊時自動關閉。
open() 基本用法#
open() 回傳一個檔案物件,使用完畢後 必須呼叫 close() 釋放資源:
1
2
3
4
5
| # 最基本的讀檔(先用傳統寫法示範,下一節就改成 with)
f = open("data.txt", "r", encoding="utf-8")
content = f.read()
f.close()
print(content)
|
try / finally → with 的演進動機#
上面那段程式有個隱藏的 bug:如果 f.read() 之後到 f.close() 之間任何地方拋出例外, close() 就永遠不會被執行。
1
2
3
4
5
6
7
8
9
10
11
12
13
| # 第一版:忘了關檔,bug!
f = open("data.txt", "r", encoding="utf-8")
content = f.read()
raise ValueError("處理時發生錯誤") # 模擬中途出錯
f.close() # ← 永遠不會執行到,檔案沒被關閉
# 執行結果:
# Traceback (most recent call last):
# File "demo.py", line 4, in <module>
# raise ValueError("處理時發生錯誤")
# ValueError: 處理時發生錯誤
#
# ↑ 程式中斷在 raise,f.close() 那一行永遠跑不到
|
1
2
3
4
5
6
7
8
| # 第二版:用 try / finally 保證關閉
f = open("data.txt", "r", encoding="utf-8")
try:
content = f.read()
raise ValueError("處理時發生錯誤") # 模擬中途出錯
finally:
f.close() # 不論 try 區塊是否拋例外,都會執行
# 例外仍會繼續向外拋,但 f 已被妥善關閉
|
這樣寫沒錯,但每次操作檔案、資料庫連線、鎖等資源時都得重複這個樣板。with 語句把「取得資源 → 使用 → 確保釋放 」這個常見模式內建到語法裡:
1
2
3
4
5
6
| # 第三版:with 自動處理 close
with open("data.txt", "r", encoding="utf-8") as f:
content = f.read()
raise ValueError("處理時發生錯誤") # 模擬中途出錯
# 離開 with 區塊時,不論是否拋例外,f.close() 都會被呼叫
# 例外仍會繼續向外拋,但 f 已被妥善關閉——效果等同第二版,但程式碼更短
|
with 之所以能做到這件事,是因為 open() 回傳的檔案物件實作了 context manager 協定 :進入 with 區塊時呼叫 __enter__() ,離開時(無論正常或例外)呼叫 __exit__() 。任何符合這個協定的物件都可以放進 with ——不只檔案,還有 threading.Lock 、資料庫連線、暫時切換工作目錄等。
從這裡開始,本章後續所有檔案範例都用 with 寫。
同時管理多個資源#
1
2
3
4
5
6
| # 一次 with,處理多個檔案
with open("input.txt", "r", encoding="utf-8") as src, \
open("output.txt", "w", encoding="utf-8") as dst:
for line in src:
dst.write(line.upper())
# src 與 dst 都會被自動關閉
|
開啟模式#
| 模式 | 說明 | 檔案不存在時 |
|---|
"r" | 讀取(預設) | 拋出 FileNotFoundError |
"w" | 寫入(會清空原內容) | 自動建立 |
"a" | 附加(append 到末尾) | 自動建立 |
"x" | 獨佔建立 | 已存在則拋 FileExistsError |
"b" | 二進位模式(搭配上述使用,如 "rb"、"wb") | — |
"+" | 讀寫模式(搭配上述使用,如 "r+"、"w+") | — |
1
2
3
4
5
6
7
8
9
10
11
12
13
| # 寫入:會把原內容清空
with open("output.txt", "w", encoding="utf-8") as f:
f.write("第一行\n")
f.write("第二行\n")
# 附加:保留原內容,加在最後
with open("output.txt", "a", encoding="utf-8") as f:
f.write("第三行(附加)\n")
# 二進位模式:讀圖片、影片等非文字資料
with open("photo.jpg", "rb") as f:
data = f.read()
print(f"檔案大小:{len(data)} bytes")
|
encoding 編碼#
讀寫文字檔時, 永遠要明確指定 encoding 。不寫的話會用「平台預設編碼」——Windows 常為 cp950 / big5、Linux / macOS 為 utf-8——同一份程式換台機器跑就可能中文亂碼或拋 UnicodeDecodeError :
1
2
3
4
5
6
| # 永遠這樣寫:明確指定 utf-8
with open("中文檔.txt", "w", encoding="utf-8") as f:
f.write("你好,世界\n")
with open("中文檔.txt", "r", encoding="utf-8") as f:
print(f.read())
|
二進位模式 "rb" / "wb" 不需要也不能指定 encoding ,因為讀寫的是原始 bytes。
讀檔的三種方式#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| # 方式 1:一次讀完整個檔案(小檔案適用)
with open("data.txt", "r", encoding="utf-8") as f:
content = f.read()
print(content)
# 方式 2:讀成串列,每行一個元素(含換行符號)
with open("data.txt", "r", encoding="utf-8") as f:
lines = f.readlines()
print(lines) # ['第一行\n', '第二行\n', ...]
# 方式 3:逐行迭代(記憶體友善,大檔案優先選這個)
with open("data.txt", "r", encoding="utf-8") as f:
for line in f:
print(line.rstrip()) # rstrip() 去掉行尾的 \n
|
自訂 context manager#
如果想讓自己寫的物件也能用在 with 裡,有兩種做法。
寫法一:class 版(__enter__ / __exit__)#
實作這兩個方法,物件就成為 context manager。下面這個 Timer 在進入時記錄起始時間,離開時印出耗時:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| import time
class Timer:
def __init__(self, label):
self.label = label
def __enter__(self):
self.start = time.perf_counter()
return self # with ... as t 會把這個值綁到 t
def __exit__(self, exc_type, exc_value, traceback):
elapsed = time.perf_counter() - self.start
print(f"[{self.label}] 耗時 {elapsed:.4f} 秒")
# 回傳 False(或 None):例外會繼續向外拋
# 回傳 True:例外會被吞掉
return False
with Timer("計算平方和"):
total = sum(i * i for i in range(1_000_000))
print(f"總和:{total}")
|
__exit__ 的三個參數在 with 區塊 正常結束時都是 None , 發生例外時帶入例外資訊 。下面示範一個資料庫風格的類別,無論查詢成功或失敗都關閉連線:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| class FakeDBConnection:
def __init__(self, dsn):
self.dsn = dsn
def __enter__(self):
print(f"連線到 {self.dsn}")
return self
def query(self, sql):
print(f"執行:{sql}")
if "DROP" in sql:
raise RuntimeError("禁止 DROP 操作")
def __exit__(self, exc_type, exc_value, traceback):
print("關閉連線")
if exc_type is not None:
print(f" 區塊內發生例外:{exc_type.__name__}:{exc_value}")
return False # 例外照常向外拋
with FakeDBConnection("localhost:5432") as db:
db.query("SELECT * FROM users")
db.query("DROP TABLE users") # 會拋例外,但連線仍會被關閉
|
寫法二:@contextlib.contextmanager 裝飾器版#
寫一整個 class 對於簡單情況有點過度。標準函式庫 contextlib 提供裝飾器,把 生成器函式 變成 context manager—— yield 之前是 __enter__ 、之後是 __exit__ :
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| from contextlib import contextmanager
import time
@contextmanager
def timer(label):
start = time.perf_counter()
try:
yield # 控制權交給 with 區塊
finally:
elapsed = time.perf_counter() - start
print(f"[{label}] 耗時 {elapsed:.4f} 秒")
with timer("計算平方和"):
total = sum(i * i for i in range(1_000_000))
|
重點 : yield 一定要包在 try / finally 裡,否則 with 區塊內若拋例外,會跳過釋放邏輯。
可以 yield 一個值出去,給 as 接:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| import os
from contextlib import contextmanager
@contextmanager
def change_dir(path):
"""暫時切換到指定目錄,離開時自動切回"""
original = os.getcwd()
os.chdir(path)
try:
yield path
finally:
os.chdir(original)
with change_dir("/tmp") as cwd:
print(f"目前位於:{cwd}")
# ... 在這裡操作 /tmp 下的檔案
print(f"已切回:{os.getcwd()}")
|
兩種寫法怎麼選?#
| 場景 | 建議寫法 |
|---|
| 簡單的「進入 → 退出」配對邏輯 | @contextmanager |
需要保留狀態、提供多個方法(像 db.query()) | class 版 |
| 需要繼承、組合其他類別 | class 版 |
| 一次性、寫起來簡短直覺 | @contextmanager |
實戰範例#
安全寫入:先寫暫存檔再改名#
避免寫到一半被中斷導致原檔損毀:
1
2
3
4
5
6
7
8
9
| import os
def safe_write(path, content):
tmp = path + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
f.write(content)
os.replace(tmp, path) # 原子性替換
safe_write("config.json", '{"version": 2}')
|
抑制特定例外: contextlib.suppress#
1
2
3
4
5
6
| from contextlib import suppress
import os
# 不存在就算了,不要當掉
with suppress(FileNotFoundError):
os.remove("maybe_not_there.tmp")
|
計算大檔案的行數(記憶體友善)#
1
2
3
4
5
| def count_lines(path):
with open(path, "r", encoding="utf-8") as f:
return sum(1 for _ in f) # 逐行迭代,不一次載入
print(count_lines("huge.log"))
|
複製檔案(二進位模式)#
1
2
3
4
5
6
| def copy_file(src, dst, chunk_size=64 * 1024):
with open(src, "rb") as fin, open(dst, "wb") as fout:
while chunk := fin.read(chunk_size):
fout.write(chunk)
copy_file("photo.jpg", "photo_backup.jpg")
|