将io.StringIO转换为io.BytesIO。

Question

原问题：我有一个StringIO对象，如何将其转换为BytesIO对象？

更新：更通用的问题是如何在Python3中将二进制（编码）类似文件的对象转换为解码的类似文件的对象？

我想到的天真方法是：

import io
sio = io.StringIO('wello horld')
bio = io.BytesIO(sio.read().encode('utf8'))
print(bio.read())  # prints b'wello horld'

有更高效且优雅的方法吗？上面的代码将所有内容读入内存，将其编码而不是以块流的方式传输数据。

例如，对于反转问题（BytesIO -> StringIO），存在一个名为io.TextIOWrapper的类，它正是这样做的（请参见此答案）。

- ShmulikA

---

“更优雅”是否包括不使用这样的大块复制来自己实现？ - Davis Herring

我希望有更好的解决方案，如果没有，那么也应该比天真的方法更好。 - ShmulikA

请注意，在原始问题中，您要求 BytesIO -> StringIO，而在更新中是 StringIO -> BytesIO。示例继续使用 BytesIO -> StringIO。 - foobarna

Answer 1

有趣的是，尽管问题似乎合理，但很难找到一个实际的原因来解释为什么我需要将StringIO转换为BytesIO。两者都基本上是缓冲区，通常只需要其中一个来进行一些额外的操作，无论是在字节还是文本上。

我可能是错的，但我认为您的问题实际上是如何在某个希望传递文本文件的代码中使用BytesIO实例。

在这种情况下，这是一个常见的问题，解决方案是使用codecs模块。

使用它的两种常见情况如下：

生成一个文件对象便于读取

In [16]: import codecs, io

In [17]: bio = io.BytesIO(b'qwe\nasd\n')

In [18]: StreamReader = codecs.getreader('utf-8')  # here you pass the encoding

In [19]: wrapper_file = StreamReader(bio)

In [20]: print(repr(wrapper_file.readline()))
'qwe\n'

In [21]: print(repr(wrapper_file.read()))
'asd\n'

In [26]: bio.seek(0)
Out[26]: 0

In [27]: for line in wrapper_file:
    ...:     print(repr(line))
    ...:
'qwe\n'
'asd\n'

组合一个文件对象以进行写入

In [28]: bio = io.BytesIO()

In [29]: StreamWriter = codecs.getwriter('utf-8')  # here you pass the encoding

In [30]: wrapper_file = StreamWriter(bio)

In [31]: print('жаба', 'цап', file=wrapper_file)

In [32]: bio.getvalue()
Out[32]: b'\xd0\xb6\xd0\xb0\xd0\xb1\xd0\xb0 \xd1\x86\xd0\xb0\xd0\xbf\n'

In [33]: repr(bio.getvalue().decode('utf-8'))
Out[33]: "'жаба цап\\n'"

- newtover

---

使用BytesIO而不是StringIO的原因之一，是为了使用upload_fileobj将内存中的文件上传到S3存储桶。更多信息请查看此处。 - cpinamtz

OutputStreamWriter是Java中请求包装器的等效物。截至2021年初，Github检索结果中有100万次使用，这正好说明了它的实用性。 - Vojtech Letal

非常有趣的字符串 'Жаба цап гадюку' =) - InvDeath

使用StringIO和BytesIO的完整示例：bytes_io = io.BytesIO(string_io.getvalue().encode()) - kellycup8

Answer 2

@foobarna answer 可以通过继承一些 io 基类来改进

import io
sio = io.StringIO('wello horld')


class BytesIOWrapper(io.BufferedReader):
    """Wrap a buffered bytes stream over TextIOBase string stream."""

    def __init__(self, text_io_buffer, encoding=None, errors=None, **kwargs):
        super(BytesIOWrapper, self).__init__(text_io_buffer, **kwargs)
        self.encoding = encoding or text_io_buffer.encoding or 'utf-8'
        self.errors = errors or text_io_buffer.errors or 'strict'

    def _encoding_call(self, method_name, *args, **kwargs):
        raw_method = getattr(self.raw, method_name)
        val = raw_method(*args, **kwargs)
        return val.encode(self.encoding, errors=self.errors)

    def read(self, size=-1):
        return self._encoding_call('read', size)

    def read1(self, size=-1):
        return self._encoding_call('read1', size)

    def peek(self, size=-1):
        return self._encoding_call('peek', size)


bio = BytesIOWrapper(sio)
print(bio.read())  # b'wello horld'

- imposeren

---

UTF8不总是单字节的。以下是错误的例子：BytesIOWrapper(io.StringIO('אבגד')).read(1)返回两个字节：b'\xd7\x90' - ShmulikA

@ShmulikA，是的，它会返回1个“字符”。为了真正返回1个字节，“intermediate”缓冲区应该被实现。 - imposeren

Answer 3

将字符流转换为字节流可能是一个普遍有用的工具，下面开始：

最初的回答：

import io

class EncodeIO(io.BufferedIOBase):
  def __init__(self,s,e='utf-8'):
    self.stream=s               # not raw, since it isn't
    self.encoding=e
    self.buf=b""                # encoded but not yet returned
  def _read(self,s): return self.stream.read(s).encode(self.encoding)
  def read(self,size=-1):
    b=self.buf
    self.buf=b""
    if size is None or size<0: return b+self._read(None)
    ret=[]
    while True:
      n=len(b)
      if size<n:
        b,self.buf=b[:size],b[size:]
        n=size
      ret.append(b)
      size-=n
      if not size: break
      b=self._read(min((size+1024)//2,size))
      if not b: break
    return b"".join(ret)
  read1=read

显然，write可以对称地定义为解码输入并将其发送到底层流，尽管这样你必须处理仅有部分字符的字节是否足够的情况。最初的回答。

- Davis Herring

---

@ShmulikA：循环永远，即使编辑过。在发布之前重写缓冲区时，我忘记了break。 - Davis Herring

Answer 4

正如一些人指出的那样，你需要自己进行编码/解码。

然而，你可以通过实现自己的 TextIOWrapper 来以优雅的方式实现 string => bytes 的转换。

以下是一个示例：

class BytesIOWrapper:
    def __init__(self, string_buffer, encoding='utf-8'):
        self.string_buffer = string_buffer
        self.encoding = encoding

    def __getattr__(self, attr):
        return getattr(self.string_buffer, attr)

    def read(self, size=-1):
        content = self.string_buffer.read(size)
        return content.encode(self.encoding)

    def write(self, b):
        content = b.decode(self.encoding)
        return self.string_buffer.write(content)

这会生成如下输出：

In [36]: bw = BytesIOWrapper(StringIO("some lengt˙˚hyÔstring in here"))

In [37]: bw.read(15)
Out[37]: b'some lengt\xcb\x99\xcb\x9ahy\xc3\x94'

In [38]: bw.tell()
Out[38]: 15

In [39]: bw.write(b'ME')
Out[39]: 2

In [40]: bw.seek(15)
Out[40]: 15

In [41]: bw.read()
Out[41]: b'MEring in here'

希望它能澄清你的思路！

- foobarna

---

read(size) 必须读取不超过 size 字节。然而，len(bw.read(15)) 是 18。 - Filip Dimitrovski

@FilipDimitrovski确实。这是因为你说“读取15个字节”，而实际上它读取的是“15个字符串字符”，其中有些字符长度为2个字节，因此出现了“18长度”。我并没有说它是完美的，但至少它没有破坏编码（通过将有效的utf-8字符分成两部分）。这只是一个示例，可以通过添加更多检查或更多方法（readline、上下文管理器等）来改进它。 - foobarna

Answer 5

我有完全相同的需求，所以我在nr.utils.io包中创建了一个EncodedStreamReader类。它还解决了从包装流中读取请求的字节数而不是字符数的问题。

$ pip install 'nr.utils.io>=0.1.0,<1.0.0'

使用示例：

import io
from nr.utils.io.readers import EncodedStreamReader
fp = EncodedStreamReader(io.StringIO('ä'), 'utf-8')
assert fp.read(1) == b'\xc3'
assert fp.read(1) == b'\xa4'

- Niklas R

Answer 6

bio 是你的例子中的 _io.BytesIO 类对象。 你使用了 2 次 read() 函数。

我提出了使用 bytes 转换和一个 read() 方法：

sio = io.StringIO('wello horld')
b = bytes(sio.read(), encoding='utf-8')
print(b)

但第二种变体应该更快：

sio = io.StringIO('wello horld')
b = sio.read().encode()
print(b)

- prosti