Python 3:使用'with'生成字节流
Python 3: Generate byte stream using 'with'
我想做的是生成一个任意长度的字节流。这样做的原因是因为我需要生成一个任意大小的文件。 (对于某些背景,我尝试流式传输一个稀疏文件,但是 had trouble when the file sizes were large,所以我尝试动态生成字节。)
我一直在阅读使对象与 with
一起工作的文档,但我正在努力正确地做到这一点。这是我在反复试验后一直在尝试的方法,但仍然无效。我想我错过了什么,但老实说我不知道是什么。
class BinaryGenerator:
def __init__(self, size_in_mbs):
self.size_in_mbs = size_in_mbs
def __enter__(self):
yield self.__bytes_generator__(self.size_in_mbs)
pass
def __exit__(self, exc_type, exc_val, exc_tb):
pass
@staticmethod
def __bytes_generator__(mega_bytes):
for i in range(mega_bytes * 1024):
yield b'\x00'
def __write_size__(size):
with open('arbitrary-{}mb.bin'.format(size), 'wb') as file:
with BinaryGenerator(size) as binary:
file.write(binary)
if __name__ == '__main__':
__write_size__(1)
我实际上会将它与 Requests Streaming Uploads 一起使用,但我正在尝试将其用作测试,看看我是否可以复制使用 [=13 打开文件时获得的二进制数据流=]
备案
a context manager 当您有一个具有初始化和清理功能的对象时使用。在示例中,它与文件对象一起使用。由于文件对象实现上下文管理器协议,退出带有文件句柄的 with
块将自动关闭文件。
with open('arbitrary-{}mb.bin'.format(size), 'wb') as fd: ...
a generator 是一个(以编程方式)生成可迭代结果的函数。可以把它想象成一个 returns 多次的函数。
至于你的问题:你可以使用生成器来实现结果流,这基本上就是它们的用途:
def bytes_generator(mega_bytes):
for i in range(mega_bytes * 1024):
yield b'\x00'
def write_size(size):
with open('arbitrary-{}mb.bin'.format(size), 'wb') as fd:
for b in bytes_generator(size):
fd.write(b)
但是,这是非常低效的,因为它一次写入一个字节。但是要修复它,您只需要修改生成器中的分块即可。
Requests 明确允许通过生成器分块上传。
def gen():
yield 'hi'
yield 'there'
requests.post('http://some.url/chunked', data=gen())
我想做的是生成一个任意长度的字节流。这样做的原因是因为我需要生成一个任意大小的文件。 (对于某些背景,我尝试流式传输一个稀疏文件,但是 had trouble when the file sizes were large,所以我尝试动态生成字节。)
我一直在阅读使对象与 with
一起工作的文档,但我正在努力正确地做到这一点。这是我在反复试验后一直在尝试的方法,但仍然无效。我想我错过了什么,但老实说我不知道是什么。
class BinaryGenerator:
def __init__(self, size_in_mbs):
self.size_in_mbs = size_in_mbs
def __enter__(self):
yield self.__bytes_generator__(self.size_in_mbs)
pass
def __exit__(self, exc_type, exc_val, exc_tb):
pass
@staticmethod
def __bytes_generator__(mega_bytes):
for i in range(mega_bytes * 1024):
yield b'\x00'
def __write_size__(size):
with open('arbitrary-{}mb.bin'.format(size), 'wb') as file:
with BinaryGenerator(size) as binary:
file.write(binary)
if __name__ == '__main__':
__write_size__(1)
我实际上会将它与 Requests Streaming Uploads 一起使用,但我正在尝试将其用作测试,看看我是否可以复制使用 [=13 打开文件时获得的二进制数据流=]
备案
a context manager 当您有一个具有初始化和清理功能的对象时使用。在示例中,它与文件对象一起使用。由于文件对象实现上下文管理器协议,退出带有文件句柄的
with
块将自动关闭文件。with open('arbitrary-{}mb.bin'.format(size), 'wb') as fd: ...
a generator 是一个(以编程方式)生成可迭代结果的函数。可以把它想象成一个 returns 多次的函数。
至于你的问题:你可以使用生成器来实现结果流,这基本上就是它们的用途:
def bytes_generator(mega_bytes):
for i in range(mega_bytes * 1024):
yield b'\x00'
def write_size(size):
with open('arbitrary-{}mb.bin'.format(size), 'wb') as fd:
for b in bytes_generator(size):
fd.write(b)
但是,这是非常低效的,因为它一次写入一个字节。但是要修复它,您只需要修改生成器中的分块即可。
Requests 明确允许通过生成器分块上传。
def gen():
yield 'hi'
yield 'there'
requests.post('http://some.url/chunked', data=gen())