前言
提到 IO,很多人脑子里只有「读文件、写文件」。但 Python 的io模块描述的是一整套流(stream)的分层模型:最底层是原始字节流,中间是带缓冲的二进制流,最上面才是把文本编码成字节的文本流。open()返回的东西,其实就是这三层拼起来的包装体。
理解这个分层,能解释很多「为什么」:为什么open()只读模式返回的是io.BufferedReader,写模式返回io.BufferedWriter;为什么文本模式下seek的偏移量是个不透明的数;为什么StringIO能直接丢给csv.reader用。
还有一个官方文档到今天仍在强调的点:open()和TextIOWrapper的默认编码是 locale 相关的(locale.getencoding()),大多数 Windows 用户的 locale 编码不是 UTF-8,于是打开 UTF-8 文件读取非 ASCII 字符就会出问题。官方因此明确建议:打开文本文件时显式指定编码。
本文讲三层 IO 模型、内存里的流(StringIO/BytesIO)、包装与重定向,以及怎么用官方提供的开关把「忘了写 encoding」这种问题主动暴露出来。示例面向 Python 3.8+。
一、IO 的三层结构
io模块把流分成三类抽象基类:
| 层 | 基类 | 处理单位 | 典型实现 |
|---|
| 原始层 | io.RawIOBase | bytes | io.FileIO |
| 缓冲层 | io.BufferedIOBase | bytes | BufferedReader、BufferedWriter、BytesIO |
| 文本层 | io.TextIOBase | str | TextIOWrapper、StringIO |
原始层直接对接系统调用,read()可能只返回部分数据,调用方要自己重试;缓冲层在上面包了一层缓冲,帮你处理「部分读写」,返回值语义变得友好;文本层再往上是编码/解码,把字节变成str。
open()的返回类型正是按这个模型决定的:文本模式返回io.TextIOWrapper,读二进制返回io.BufferedReader,写二进制返回io.BufferedWriter。
buffering参数控制缓冲策略:
# 适用于 Python 3.8+
open(path, "rb", buffering=0) # 关闭缓冲(只允许二进制模式)
open(path, "w", buffering=1) # 行缓冲(只在文本写入模式下可用)
open(path, "rb", buffering=65536) # 指定固定大小的块缓冲传buffering=0只有在二进制模式才合法,文本模式下会报错——因为文本层本身就依赖缓冲。
二、内存里的流:StringIO 与 BytesIO
不是所有「文件」都得落在磁盘上。当某个函数接受一个文件对象时,你完全可以用内存流顶替:
# 适用于 Python 3.8+
import io
buf = io.StringIO("第一行\n第二行\n")
print(buf.readline()) # 第一行\n
print(buf.read()) # 第二行\n
buf.seek(0)
print(buf.getvalue()) # 拿到全部内容(不移动当前位置)io.StringIO(initial_value='', newline='\n')是文本内存流,io.BytesIO(initial_bytes=b'')是二进制内存流:
# 适用于 Python 3.8+
import io
b = io.BytesIO()
b.write(b"\x00\x01\x02")
print(b.getvalue()) # b'\x00\x01\x02'
b.seek(0)
print(b.read(2)) # b'\x00\x01'内存流最大的价值是让「写文件」的接口变成可测试的纯内存操作。例如csv.writer接受任何带write()方法的对象,就可以把它接到StringIO上,在内存里生成 CSV 再检查内容,完全不碰磁盘:
# 适用于 Python 3.8+
import csv
import io
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(["姓名", "分数"])
writer.writerow(["张三", "90"])
print(buf.getvalue())这里要提醒一句:StringIO和真正的文件对象并不完全等价。它没有文件描述符,没有fileno(),也不能被select之类的系统接口使用。用它替换文件对象时,先确认下游只调用了read/write/seek/getvalue这类通用方法。
三、包装与重定向
io.TextIOWrapper可以把一个二进制流包成文本流。它的完整签名是:
io.TextIOWrapper(buffer, encoding=None, errors=None, newline=None,
line_buffering=False, write_through=False)一个实用场景是:先用二进制方式拿到数据,再按指定编码解包。这比「二进制读出来自己 decode 切行」要省事得多。
另一个常见需求是把print的输出临时改道。contextlib.redirect_stdout就是干这个的:
# 适用于 Python 3.8+
import io
from contextlib import redirect_stdout
buf = io.StringIO()
with redirect_stdout(buf):
print("这段字被重定向了")
print("捕获到:", buf.getvalue().strip())redirect_stdout只在with块内生效,退出后sys.stdout自动还原。这类「配对操作」用上下文管理器写,比手写sys.stdout = buf再手动还原来得安全——后者的还原语句一旦被异常跳过,整个进程的输出就都跑到缓冲区里去了。
让「忘了写 encoding」主动报警
Python 3.10 起提供了两个开关,可以在代码用到默认 locale 编码时发出EncodingWarning:
# 适用于 Python 3.10+
# 运行时打开警告
python -X warn_default_encoding your_script.py# 适用于 Python 3.10+
# 或者用环境变量
PYTHONWARNDEFAULTENCODING=1 python your_script.py同一版本还加入了encoding="locale"这个特殊取值,用来「显式地表示我要用当前 locale 编码」。另外,io.text_encoding()是给库作者用的:在自己的 API 里接收encoding=None时,可以借它让调用方自动收到EncodingWarning。
实战:把文本处理函数改成可测试的
下面这个函数不直接写磁盘,而是接受一个「类文件对象」,这样既能写文件也能写内存,测试起来方便:
# 适用于 Python 3.8+
import io
def dump_pairs(pairs, stream):
"""把 (键, 值) 序列写成 key=value 每行一条。"""
for key, value in pairs:
stream.write(f"{key}={value}\n")
if __name__ == "__main__":
# 真实场景:写文件
with open("out.ini", "w", encoding="utf-8") as f:
dump_pairs([("host", "localhost"), ("port", 8080)], f)
# 测试场景:写内存
buf = io.StringIO()
dump_pairs([("host", "localhost"), ("port", 8080)], buf)
assert buf.getvalue() == "host=localhost\nport=8080\n"这种「面向流编程」的写法是 IO 分层的直接收益:同一份逻辑,换个流对象就能跑在不同介质上。
常见坑点
1. 把StringIO当真实文件用
❌io.StringIO().fileno(),抛io.UnsupportedOperation。 ✅ 需要文件描述符就用tempfile.TemporaryFile();只是做数据缓冲就用StringIO。
2.BytesIO和StringIO混用
❌ 对BytesIO调write("文本"),抛TypeError: a bytes-like object is required。 ✅ 二进制流写bytes,文本流写str,两者不要串。
3.getvalue()与read()混淆
❌ 想取全部内容却调了read(),把游标推到了末尾,后续读取全空。 ✅ 只取内容不动游标用getvalue()。
4. 文本模式关缓冲
❌open(path, "w", buffering=0),抛ValueError: can't have unbuffered text I/O。 ✅ 关闭缓冲只在二进制模式可用,文本模式至少要有缓冲。
5. 手写sys.stdout替换却不还原
❌sys.stdout = buf之后中途抛异常,还原语句没执行,整个程序输出消失。 ✅ 用with redirect_stdout(buf):,退出时自动还原。
6. 忘了写 encoding 却在 Windows 上测不出问题
❌open("a.json")在中文 Windows 上用 cp936,读 UTF-8 内容抛UnicodeDecodeError。 ✅ 始终写encoding="utf-8";排查阶段用-X warn_default_encoding打开警告。
7. 原始流只读一次就当读完了
❌ 以为RawIOBase.read()一次就能拿到全部数据,结果拿到的是部分字节。 ✅ 用缓冲层(open()默认就是),或自己循环读到空为止。
8. 缓冲没刷新就让别人读文件
❌ 写完没关也没flush(),另一个进程读到的还是旧内容。 ✅ 用with退出自动 flush,或显式调用f.flush()。
总结
| 需求 | 用什么 | 关键点 |
|---|
| 处理文本 | open(..., encoding="utf-8") | 默认是 locale 编码,跨平台不一致 |
| 处理字节 | open(..., "rb") | 不能传encoding |
| 内存文本流 | io.StringIO | 有getvalue(),没有fileno() |
| 内存字节流 | io.BytesIO | 只能写bytes |
| 二进制转文本 | io.TextIOWrapper | 需要自己指定编码 |
| 重定向输出 | contextlib.redirect_stdout | 上下文管理器,自动还原 |
| 排查编码 | -X warn_default_encoding | Python 3.10+,触发EncodingWarning |
IO 这套分层设计的核心思想是「把编码、缓冲、系统调用拆成可以单独替换的层」。理解它以后,open()返回什么类型、seek为什么在文本模式下受限、内存流为什么能顶替文件,这些疑问都会变成同一个答案。