Python FTP“块”迭代器（不将整个文件加载到内存中）

Question

关于检索 FTP 文件并将其写入流（如字符串缓冲区或随后可以迭代的文件）的堆栈溢出有几个答案。

如：Read a file in buffer from FTP python

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

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

I.E.而不是：

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 文件写入磁盘或内存。

Answer 1

您必须将 retrbinary 调用放在另一个线程中，并将回调提要块放入迭代器：

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