ioutils -增强输入/输出功能

模块 ioutils 实现许多帮助器类和函数,这些类和函数在以各种方式处理输入、输出和字节流时非常有用。

假脱机临时文件

假脱机临时文件是类似文件的对象,它们一开始映射到内存中的对象,但一旦达到某个(可配置)阈值,就会自动转存到临时文件。遗憾的是,Python中内置的SpooledTemporaryFile类并没有实现某些常见类(如StringIO)所实现的API。SpooledTemporaryFile还将其内存中的所有文件假脱机为cStringIO实例。CStringIO实例不能被深度复制,它们也不能与Zip库一起工作。这一点,再加上不兼容的API,使得它在一些用例中毫无用处。

为了解决这一问题,同时仍然获得真正的假脱机文件类对象的内存节省和可用性,已经实现了两个具有兼容API的定制类。

SpooledBytesIO

class boltons.ioutils.SpooledBytesIO(max_size=5000000, dir=None)[源代码]

SpooledBytesIO是一个假脱机的类似文件的对象,它只接受字节。在Python2.x上,这意味着‘str’类型;在Python3.x上,这意味着‘bytes’类型。字节被完全按照给定的方式写入和检索,但如果写入的是字节以外的内容,则会引发类型错误。

示例::

>>> from boltons import ioutils
>>> with ioutils.SpooledBytesIO() as f:
...     f.write(b"Happy IO")
...     _ = f.seek(0)
...     isinstance(f.getvalue(), bytes)
True

SpooledStringIO

class boltons.ioutils.SpooledStringIO(*args, **kwargs)[源代码]

SpooledStringIO是一个假脱机的类似文件的对象,它只接受Unicode值。在Python2.x上,这意味着‘unicode’类型,而在Python3.x上,这意味着‘str’类型。值被接受为Unicode,然后被强制到UTF-8编码字节中进行存储。检索时,这些值以Unicode的形式返回。

示例::

>>> from boltons import ioutils
>>> with ioutils.SpooledStringIO() as f:
...     f.write(u"— Hey, an emdash!")
...     _ = f.seek(0)
...     isinstance(f.read(), str)
True

实例

在旧的Python代码中过度使用StringIO的情况并不少见。如果想要减少内存开销,SpooledTemporaryFile将是一个很好的替代品,但不幸的是,它的API差别太大了。这是一个很好的候选人 SpooledBytesIO 因为它是与API兼容的,因此可以用作插入式替代。

旧代码::

flo = StringIO()
flo.write(gigantic_string)

更新::

from boltons.ioutils import SpooledBytesIO

flo = SpooledBytesIO()
flo.write(gigantic_string)

另一个很好的用例是从某个远程位置下载文件。如果文件很小,将其保存在内存中是很好的,但将一个大文件写入内存可能会使服务器非常恼火。如果下载的文件恰好是一个压缩文件,那么情况就更糟了。您不能使用普通的SpooledTemporaryFile,因为它不兼容。一个 SpooledBytesIO 实例是一个很好的替代方案。下面是一个使用请求库下载压缩文件的简单示例:

from zipfile import ZipFile

import requests
from boltons import ioutils

# Using a context manager with stream=True ensures the connection is closed. See:
# http://docs.python-requests.org/en/master/user/advanced/#body-content-workflow
with requests.get("http://127.0.0.1/test_file.zip", stream=True) as r:
    if r.status_code == 200:
        with ioutils.SpooledBytesIO() as flo:
            for chunk in r.iter_content(chunk_size=64000):
                flo.write(chunk)

            flo.seek(0)

            zip_doc = ZipFile(flo)

            # Print all the files in the zip
            print(zip_doc.namelist())

多个文件

MultiFileReader

class boltons.ioutils.MultiFileReader(*fileobjs)[源代码]

获取打开的文件或类似文件的对象的列表,并提供一个连续读取它们的接口。喜欢 itertools.chain() ,但用于读取文件。

>>> mfr = MultiFileReader(BytesIO(b'ab'), BytesIO(b'cd'), BytesIO(b'e'))
>>> mfr.read(3).decode('ascii')
u'abc'
>>> mfr.read(3).decode('ascii')
u'de'

构造函数接受您交给它的任意数量的fileobjs,并将在非类文件对象上引发TypeError。当类似文件的对象是字节处理对象和文本处理对象(例如,BytesIO和StringIO)的混合时,会引发ValueError。