【Python Cookbook】文件与 IO（二）

6.字符串的 I/O 操作

你想使用操作类文件对象的程序来操作文本或二进制字符串。

使用 io.StringIO() 和 io.BytesIO() 类来创建类文件对象操作字符串数据。比如：

>>> s = io.StringIO()
>>> s.write('Hello World\n')
12
>>> print('This is a test', file=s)
15
>>> # Get all of the data written so far
>>> s.getvalue()
'Hello World\nThis is a test\n'
>>>>>> # Wrap a file interface around an existing string
>>> s = io.StringIO('Hello\nWorld\n')
>>> s.read(4)
'Hell'
>>> s.read()
'o\nWorld\n'
>>>

io.StringIO 只能用于文本。如果你要操作二进制数据，要使用 io.BytesIO 类来代替。比如：

>>> s = io.BytesIO()
>>> s.write(b'binary data')
>>> s.getvalue()
b'binary data'
>>>

当你想模拟一个普通文件的时候，StringIO 和 BytesIO 类是很有用的。比如，在单元测试中，你可以使用 StringIO 来创建一个包含测试数据的类文件对象，这个对象可以被传给某个参数为普通文件对象的函数。

需要注意的是， StringIO 和 BytesIO 实例并没有正确的整数类型的文件描述符。因此，它们不能在那些需要使用真实的系统级文件，如文件，管道或者是套接字的程序中使用。

🚀 在博主的另一篇博客《解析 io.StringIO 与 io.BytesIO》中有更为详细的介绍。

7.读写压缩文件

你想读写一个 gzip 或 bz2 格式的压缩文件。

gzip 和 bz2 模块可以很容易的处理这些文件。 两个模块都为 open() 函数提供了另外的实现来解决这个问题。比如，为了以文本形式读取压缩文件，可以这样做：

# gzip compression
import gzip
with gzip.open('somefile.gz', 'rt') as f:text = f.read()# bz2 compression
import bz2
with bz2.open('somefile.bz2', 'rt') as f:text = f.read()

类似的，为了写入压缩数据，可以这样做：

# gzip compression
import gzip
with gzip.open('somefile.gz', 'wt') as f:f.write(text)# bz2 compression
import bz2
with bz2.open('somefile.bz2', 'wt') as f:f.write(text)

如上，所有的 I/O 操作都使用文本模式并执行 Unicode 的编码/解码。类似的，如果你想操作二进制数据，使用 rb 或者 wb 文件模式即可。

大部分情况下读写压缩数据都是很简单的。但是要注意的是选择一个正确的文件模式是非常重要的。如果你不指定模式，那么默认的就是二进制模式，如果这时候程序想要接受的是文本数据，那么就会出错。gzip.open() 和 bz2.open() 接受跟内置的 open() 函数一样的参数，包括 encoding，errors，newline 等等。

当写入压缩数据时，可以使用 compresslevel 这个可选的关键字参数来指定一个压缩级别。比如：

with gzip.open('somefile.gz', 'wt', compresslevel=5) as f:f.write(text)

默认的等级是 $9$，也是最高的压缩等级。等级越低性能越好，但是数据压缩程度也越低。

最后一点， gzip.open() 和 bz2.open() 还有一个很少被知道的特性，它们可以作用在一个已存在并以二进制模式打开的文件上。比如，下面代码是可行的：

import gzip
f = open('somefile.gz', 'rb')
with gzip.open(f, 'rt') as g:text = g.read()

这样就允许 gzip 和 bz2 模块可以工作在许多类文件对象上，比如套接字，管道和内存中文件等。

8.固定大小记录的文件迭代（⭐⭐）

你想在一个固定长度记录或者数据块的集合上迭代，而不是在一个文件中一行一行的迭代。

通过下面这个小技巧使用 iter 和 functools.partial() 函数：

from functools import partialRECORD_SIZE = 32with open('somefile.data', 'rb') as f:records = iter(partial(f.read, RECORD_SIZE), b'')for r in records:...

• partial(f.read, RECORD_SIZE):
  • functools.partial 创建一个新的函数，这个新函数会固定 f.read 的第一个参数为 RECORD_SIZE（即 32）。
  • 相当于每次调用 partial(f.read, RECORD_SIZE)() 都会执行 f.read(32)，即从文件中读取 32 字节的数据。
• iter(callable, sentinel):
  • iter 不仅可以用于可迭代对象，还可以接受一个可调用对象（callable）和一个哨兵值（sentinel）。
  • 它会重复调用 callable，直到返回 sentinel 为止，此时迭代停止。
  • 在这里，callable 是 partial(f.read, RECORD_SIZE)，sentinel 是 b''（空字节串）。
  • 因此，iter 会不断调用 f.read(32)，直到返回空字节串（表示文件读取完毕），然后停止迭代。
• for r in records::
  • records 是一个迭代器，每次迭代会返回一个最多 RECORD_SIZE 字节的记录（r）。
  • 当文件读取完毕时，f.read(32) 返回 b''，迭代终止。

这个例子中的 records 对象是一个可迭代对象，它会不断的产生固定大小的数据块，直到文件末尾。要注意的是如果总记录大小不是块大小的整数倍的话，最后一个返回元素的字节数会比期望值少。

iter() 函数有一个鲜为人知的特性就是，如果你给它传递一个可调用对象和一个标记值，它会创建一个迭代器。这个迭代器会一直调用传入的可调用对象直到它返回标记值为止，这时候迭代终止。

在例子中， functools.partial 用来创建一个每次被调用时从文件中读取固定数目字节的可调用对象。标记值 b'' 就是当到达文件结尾时的返回值。

最后再提一点，上面的例子中的文件是以二进制模式打开的。如果是读取固定大小的记录，这通常是最普遍的情况。而对于文本文件，一行一行的读取（默认的迭代行为）更普遍点。

🚀 假设 somefile.data 的内容是 b'HelloWorld' * 10（即重复 10 次的 b'HelloWorld'，共 100 字节），并且 RECORD_SIZE = 32。

文件内容

b'HelloWorldHelloWorldHelloWorldHelloWorldHelloWorldHelloWorldHelloWorldHelloWorldHelloWorldHelloWorld'

运行代码

from functools import partialRECORD_SIZE = 32with open('somefile.data', 'rb') as f:records = iter(partial(f.read, RECORD_SIZE), b'')for i, r in enumerate(records, 1):print(f"Record {i}: {r}")

输出

Record 1: b'HelloWorldHelloWorldHelloWorldHello'
Record 2: b'WorldHelloWorldHelloWorldHelloWorld'
Record 3: b'HelloWorldHelloWorldHelloWorldHello'
Record 4: b'World'

说明

• 文件共100字节，每次读取32字节：
  • 第1次读取：32字节（b'HelloWorldHelloWorldHelloWorldHello'）
  • 第2次读取：32字节（b'WorldHelloWorldHelloWorldHelloWorld'）
  • 第3次读取：32字节（b'HelloWorldHelloWorldHelloWorldHello'）
  • 第4次读取：剩余4字节（b'World'）
  • 第5次读取：返回 b''，迭代终止。

适用场景

这种方法非常适合处理固定大小的记录文件，例如：

• 二进制文件格式（如数据库文件、图像文件等）。
• 网络协议数据包（固定大小的数据块）。
• 任何需要分块处理的流式数据。