Python：将内容写入打开的流

Question

我正在使用一个 API，该 API 将打开的二进制文件作为参数，然后对其执行阻塞读取，直到 EOF。

而不是打开现有文件（io.open 模式'rb'），我想向它传递一个流，我将计算/构造的内容写入其中 - 实际上我想要一些概念上是输出输出的单向管道通过可与打开文件互换的输入流。

我查看了 BufferedRWPair，但我能找到的几个例子都违反了不要在输入端和输出端使用相同对象的警告。

如果有人有合适的例子或更好的建议，欢迎！

我在这里查看了基于 cmets 的 BufferedRandom，但显然我做错了什么...... import io buf = io.BufferedRandom(io.BytesIO()) buf.write("a") buf.write("b") buf.flush() while True: print "reading" a = buf.read(1024) if not a: break print "read: {}".format(a) buf.close() 这在第一次读取后退出

更新 这个公认的混乱示例显示了解决方案，必须保持独立的读写位置

import io
buf = io.BufferedRandom(io.BytesIO())
read = 0
wrote = 0
buf.seek(wrote)
wrote += buf.write(b"a")
wrote += buf.write(b"b")
buf.seek(read)
data = buf.read(1)
read += len(data)
buf.seek(wrote)
wrote += buf.write(b"c")
print "read: {}".format(data)
buf.seek(read)
data = buf.read(512)
read += len(data)
wrote += buf.write(b"d")

buf.seek(wrote)
wrote += buf.write(b"efghihjlmnop")
while data:
    print "read: {}".format(data)
    buf.seek(read)
    data = buf.read(1024)
    read += len(data)
buf.close()

Answer 1

评论：...允许我在流中交错读取和写入，而无需...自己管理当前的读取和写入位置。

这是io.BufferedRandom的行为。
但是你可以把逻辑封装成一个自己的class StreamRW(io.BufferedRandom),
例如：

class StreamRW(io.BufferedRandom):
    def __init__(self, raw):
        super().__init__(raw)
        self.seek(0)

    def read(self, size=1):
        super().seek(self.read_offset)
        data = super().read(size)
        self.read_offset = self.tell()
        return data

    def write(self, data):
        super().seek(self.write_offset)
        written = super().write(data)
        self.write_offset = self.tell()
        return written

    def seek(self, offset):
        super().seek(offset)
        self.read_offset = self.write_offset = self.tell()

#Usage:
buf = StreamRW(io.BytesIO())
...

更多代码如下，但没有buf.seek(0)！

---

你必须使用buf.seek(0)来回退文件位置。
注意：我必须使用binary prefix b""！
这对我有用：

import io
buf = io.BufferedRandom(io.BytesIO())
buf.write(b"a")
buf.write(b"b")
buf.seek(0)

while True:
    print "reading"
    a = buf.read(1024)
    if not a: break
    print "read: {}".format(a)
buf.close()

输出：
阅读：b'ab'

用 Python 测试：3.4.2 和 2.7.9