Python标注库之IO模块

io 模块简介

io 模块是 Python 标准库中的一个核心模块，提供了 Python 对 I/O 操作的基本支持。它支持各种文件和流的读写操作，并且为不同类型的 I/O 操作提供了统一的接口。io 模块是 Python 3 引入的，用于替代 Python 2 中的 file 对象和一些旧的 I/O 相关模块。

主要功能

• 文件和流的读写：提供了文件操作的基本功能。
• 内存缓冲：支持在内存中进行文件操作。
• 文本和二进制流：支持处理文本流和二进制流。
• 自定义 I/O：支持自定义实现输入输出流。

使用场景

• 文件操作：io 模块用于处理文件和流的读写操作。
• 内存操作：使用 BytesIO 和 StringIO 在内存中处理数据，适用于测试和处理临时数据。
• 自定义流：通过继承 IOBase 类，可以实现自定义的 I/O 行为和数据流处理。

io 模块的使用

核心类

• IOBase:这是所有流对象的抽象基类。它不提供具体实现，但定义了一些基础接口。
• RawIOBase:继承自IOBase，用于处理原始二进制流。直接处理字节，不进行任何编码或解码。
• BufferedIOBase:继承自IOBase，用于缓冲二进制流。这类流支持更高效的读写操作。
• TextIOBase:继承自IOBase，用于文本流。它处理字符串，并支持编码和解码。

io.IOBase：io 模块中所有流的基类，定义了常用的 I/O 操作接口。

方法：

• read(size=-1)：读取数据。
• write(data)：写入数据。
• flush()：刷新缓冲区。
• seek(offset, whence)：设置当前位置。
• tell()：返回当前位置。

import io

class CustomIO(io.IOBase):
    def __init__(self, initial_data):
        self.data = initial_data
        self.position = 0

    def read(self, size=-1):
        if size == -1:
            size = len(self.data) - self.position
        result = self.data[self.position:self.position + size]
        self.position += size
        return result

    def write(self, data):
        self.data += data

    def seek(self, offset, whence=0):
        if whence == 0:
            self.position = offset
        elif whence == 1:
            self.position += offset
        elif whence == 2:
            self.position = len(self.data) + offset

custom_io = CustomIO('Hello, World!')
print(custom_io.read(5))  # 输出 'Hello'
custom_io.write(' Python!')
custom_io.seek(0)
print(custom_io.read())  # 输出 'Hello, World! Python!'

具体实现类

• FileIO:继承自RawIOBase，用于读取和写入文件的原始字节流。
• BytesIO:继承自BufferedIOBase，用于在内存中读写二进制数据。适合于需要在内存中处理二进制数据的场景。
• BufferedReader:继承自BufferedIOBase，用于缓冲读取二进制数据。
• BufferedWriter:继承自BufferedIOBase，用于缓冲写入二进制数据。
• BufferedRandom:继承自BufferedIOBase，支持可读可写的缓冲二进制流。
• StringIO:继承自TextIOBase，用于在内存中读写文本数据。适合于需要在内存中处理字符串的场景。
• TextIOWrapper:继承自TextIOBase，用于对缓冲二进制流进行编码和解码，通常用于将 BufferedIOBase 的实例转换为文本流。

io.StringIO：在内存中读写文本数据的流。

方法：

• read(size=-1)：读取缓冲区内容。
• write(string)：写入字符串数据。
• seek(offset, whence)：设置当前位置。
• tell()：返回当前位置。

import io

# 创建一个 StringIO 对象
s = io.StringIO()

# 写入文本
s.write('Hello, ')
s.write('World!\n')

# 获取当前内容
content = s.getvalue()
print(content)  # 输出: Hello, World!

# 读取内容
s.seek(0)  # 移动到文件开头
print(s.read())  # 输出: Hello, World!

# 关闭对象
s.close()

io.BytesIO：在内存中读写二进制数据的流。

方法：

• read(size=-1)：读取缓冲区内容。
• write(b)：写入字节数据。
• seek(offset, whence)：设置当前位置。
• tell()：返回当前位置。

import io

# 创建一个 BytesIO 对象
b = io.BytesIO()

# 写入二进制数据
b.write(b'Hello, ')
b.write(b'World!\n')

# 获取当前内容
content = b.getvalue()
print(content)  # 输出: b'Hello, World!'

# 读取内容
b.seek(0)  # 移动到文件开头
print(b.read())  # 输出: b'Hello, World!'

# 关闭对象
b.close()

io.TextIOWrapper：提供对文本文件的读取和写入。

方法：

• read(size=-1)：读取文件内容，size 参数指定要读取的字节数。
• write(string)：将字符串写入文件。
• flush()：刷新文件的内部缓冲区。
• seek(offset, whence)：设置文件当前的位置。
• tell()：返回文件当前的位置。

import io

# 打开一个二进制文件
with open('binary_file.bin', 'rb') as binary_file:
    # 将二进制流包装为文本流
    text_file = io.TextIOWrapper(binary_file, encoding='utf-8')
    
    # 读取文本内容
    content = text_file.read()
    print(content)
    
    # 关闭文本流
    text_file.detach()  # 断开与二进制流的连接

io.BufferedReader 和 io.BufferedWriter：提供对二进制文件的缓冲读写操作。

方法：

• read(size=-1)：读取指定字节数的数据。
• write(b)：写入字节数据。
• flush()：刷新缓冲区。

import io

# 打开一个文件进行读取
with open('large_file.bin', 'rb') as raw_file:
    buffered_reader = io.BufferedReader(raw_file)
    
    # 读取数据
    chunk = buffered_reader.read(1024)
    while chunk:
        print(chunk)
        chunk = buffered_reader.read(1024)

# 打开一个文件进行写入
with open('output_file.bin', 'wb') as raw_file:
    buffered_writer = io.BufferedWriter(raw_file)
    
    # 写入数据
    data = b'This is some binary data.'
    buffered_writer.write(data)
    buffered_writer.flush()  # 确保数据写入文件

io.FileIO

FileIO 类用于处理原始的二进制文件操作。

import io

# 打开一个文件进行读取
with io.FileIO('example.txt', 'r') as file:
    content = file.read()
    print(content.decode('utf-8'))  # 解码为文本

# 打开一个文件进行写入
with io.FileIO('example.txt', 'w') as file:
    file.write('Hello, World!'.encode('utf-8'))

常用函数

io.open(file, mode=’r’, buffering=-1, encoding=None, errors=None, newline=None, closefd=True, opener=None):

• file：要打开的文件名。
• mode：打开文件的模式，如 ‘r’（读取）、’w’（写入）、’b’（二进制）等。
• buffering：缓冲策略，默认为 -1（自动选择缓冲策略）。
• encoding：文本编码，默认为 None（使用系统默认编码）。
• errors：指定错误处理方案，如 ‘ignore’ 或 ‘replace’。
• newline：控制如何处理换行符，默认为 None。

用于打开文件并返回相应的文件对象。open 是 io 模块中最常用的函数之一，允许指定文件的打开模式（如读、写、追加等）和编码方式。

示例：

import io

# 打开文件进行读取
with io.open('file.txt', 'r', encoding='utf-8') as f:
    content = f.read()
    print(content)

注意事项

• 模式：在打开文件时，确保选择正确的模式（文本模式还是二进制模式）。
• 编码：处理文本数据时，确保使用正确的编码，避免数据损坏。
• 资源管理：使用 with 语句确保文件和流被正确关闭，避免资源泄漏。