Python使用TextIOWrapper以二进制模式正确读写CSV的实现与问题排查

问题背景 #

参考StackOverflow两个相关技术问题《How to write a csv file in binary mode?》《Python: reading from binary zipfile into csv》的解决方案调试测试代码时，始终无法明确该场景下的文件正确处理方式。

核心诉求 #

• 原始写入的b'\x01'这类bytes类型值，最终从文件读回时需保留bytes类型，当前实现读回的值均为字符串类型。
• 解决树莓派OS（Linux环境）Python 3.7版本上，测试代码运行到第8行抛出AttributeError: 'file' object has no attribute 'readable'的问题，相同代码在Windows 10 Python 3.9环境下可无错运行。
• 明确该场景下文件流的正确处理方法。

测试代码 #

import io
import csv

test_byte = b'\x01'
print ("original item value is %s of type %s" % (test_byte, type(test_byte)))
csv_data_dict = [{"preset": "item", "value": test_byte}]

with io.TextIOWrapper(open("test.csv", 'wb'), encoding = 'utf-8', newline = '') as csvfile:
    fieldnames = ["preset", "value"]
    writer = csv.DictWriter(csvfile, fieldnames = fieldnames)
    writer.writeheader()
    for i in range(len(csv_data_dict)):
        writer.writerow(csv_data_dict[i])

with io.TextIOWrapper(open("test.csv", 'rb'), encoding = 'utf-8', newline = '') as csvfile:
    reader = csv.DictReader(csvfile)
    for row in reader:
        if row["preset"] == "item":
            print ("item value returned as %s of type %s" % (row["value"], type(row["value"]))) # 读回的值为字符串类型

Windows 10 环境运行输出 #

original item value is b'\x01' of type <class 'bytes'>
item value returned as b'\x01' of type <class 'str'>

问题原因分析 #

1. 跨版本报错原因：手动将open("test.csv", 'wb'/'rb')返回的原始文件对象传给io.TextIOWrapper的用法冗余且不兼容。Python 3.7的Linux环境下，open()默认返回的是无缓冲的原始文件对象，不具备TextIOWrapper要求的完整缓冲IO接口方法，因此触发属性缺失报错；而Python 3.9的Windows环境下open()默认返回带缓冲的IO对象，可适配手动包装逻辑，因此不会报错。
2. bytes读回变字符串的原因：CSV格式本身是纯文本规范，csv模块仅支持文本流读写，所有写入的非字符串类型值都会被自动转成字符串形式，最终读回的必然是字符串类型，不可能直接保留原生bytes类型。

解决方案 #

1. 修正文件流用法，解决跨版本报错 #

无需手动嵌套open()和TextIOWrapper，直接使用open()内置的encoding参数即可自动生成兼容所有Python 3版本的文本流，修正后的基础代码逻辑：

import csv

test_byte = b'\x01'
print ("原始值 %s 类型为 %s" % (test_byte, type(test_byte)))
csv_data_dict = [{"preset": "item", "value": test_byte}]

# 写入逻辑：直接在open中指定编码，无需手动套TextIOWrapper
with open("test.csv", 'w', encoding='utf-8', newline='') as csvfile:
    fieldnames = ["preset", "value"]
    writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
    writer.writeheader()
    for row in csv_data_dict:
        writer.writerow(row)

# 读取逻辑同理
with open("test.csv", 'r', encoding='utf-8', newline='') as csvfile:
    reader = csv.DictReader(csvfile)
    for row in reader:
        if row["preset"] == "item":
            print ("读回值 %s 类型为 %s" % (row["value"], type(row["value"])))

2. 实现bytes类型值的持久化与还原 #

由于CSV本身无法存储原生bytes，需要先将bytes序列化为可打印的字符串格式（如base64编码）写入CSV，读取时再反序列化回bytes：

import csv
import base64

test_byte = b'\x01'
print ("原始值 %s 类型为 %s" % (test_byte, type(test_byte)))

# 写入前将bytes转为base64编码字符串
encoded_byte = base64.b64encode(test_byte).decode('ascii')
csv_data_dict = [{"preset": "item", "value": encoded_byte}]

with open("test.csv", 'w', encoding='utf-8', newline='') as csvfile:
    fieldnames = ["preset", "value"]
    writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
    writer.writeheader()
    for row in csv_data_dict:
        writer.writerow(row)

with open("test.csv", 'r', encoding='utf-8', newline='') as csvfile:
    reader = csv.DictReader(csvfile)
    for row in reader:
        if row["preset"] == "item":
            # 读取时将base64字符串解码回bytes
            decoded_byte = base64.b64decode(row["value"])
            print ("还原后的值 %s 类型为 %s" % (decoded_byte, type(decoded_byte)))

运行上述代码后，读回的decoded_byte将和原始值完全一致，类型为bytes。

内容的提问来源于stack exchange，提问作者secarica