Python FTP «chunk» iterator (без загрузки всего файла в память)

Question

Существует несколько ответов на переполнение стека по извлечению файла FTP и запись его в поток, такой как строковый буфер или файл, который затем можно повторить.

Такие, как: Read a file in buffer from FTP python

Однако эти решения предполагают загрузку всего файла в память или загрузить его на диск перед началом обработки содержимого.

У меня недостаточно памяти для хранения всего файла, и у меня нет доступа к диску. Это можно сделать, обработав данные в функции обратного вызова, но Я хочу знать, можно ли обернуть ftp-код в какой-то магии, которая возвращает итератор, а не перегружает мой код обратными вызовами.

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) 

И (в отличие от существующих ответов) Я хочу сделать это без написания весь передачи файлов файл на диске или в памяти, прежде чем итерация на нем.

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