Python FTP“块”迭代器(无需将整个文件加载到内存中)

关于获取FTP文件并将其写入流(例如字符串缓冲区或文件,然后可以对其进行迭代)的堆栈溢出问题,有几个答案.如：Read a file in buffer from FTP python但是,这些解决方案涉及在开始处理内容之前将整个文件加载到内存...

关于获取FTP文件并将其写入流(例如字符串缓冲区或文件,然后可以对其进行迭代)的堆栈溢出问题,有几个答案.

如：Read a file in buffer from FTP python

但是,这些解决方案涉及在开始处理内容之前将整个文件加载到内存中或将其下载到磁盘.

我没有足够的内存来缓冲整个文件,并且无法访问该磁盘.这可以通过处理回调函数中的数据来完成,但是我想知道是否有可能将ftp代码包装在某种魔术中,该魔术返回一个迭代器,而不是在回调中添加代码.

即而不是：

def get_ftp_data(handle_chunk):
    ...
    ftp.login('uesr', 'password') # authentication required
    ftp.retrbinary('RETR etc', handle_chunk)
    ...

get_ftp_data(do_stuff_to_chunk)

我想要：

for chunk in get_ftp_data():
    do_stuff_to_chunk(chunk)

并且(与现有答案不同),我希望在迭代之前不将整个ftp文件写入磁盘或内存中而执行此操作.

解决方法:

您必须将retrbinary调用放在另一个线程中,并将回调feed块传递给迭代器：

import threading, Queue

def ftp_chunk_iterator(FTP, command):
    # Set maxsize to limit the number of chunks kept in memory at once.
    queue = Queue.Queue(maxsize=some_appropriate_size)

    def ftp_thread_target():
        FTP.retrbinary(command, callback=queue.put)
        queue.put(None)

    ftp_thread = threading.Thread(target=ftp_thread_target)
    ftp_thread.start()

    while True:
        chunk = queue.get()
        if chunk is not None:
            yield chunk
        else:
            return

如果您不能使用线程,则最好的办法是将回调编写为协程：

from contextlib import closing


def process_chunks():
    while True:
        try:
            chunk = yield
        except GeneratorExit:
            finish_up()
            return
        else:
            do_whatever_with(chunk)

with closing(process_chunks()) as coroutine:

    # Get the coroutine to the first yield
    coroutine.next()

    FTP.retrbinary(command, callback=coroutine.send)
# coroutine.close() #  called by exiting the block