Python FTP „chunk“ Iterator (ohne die gesamte Datei in den Speicher geladen)

Question

Es gibt mehrere Antworten auf Stapelüberlauf über eine FTP-Datei abgerufen und in einen Strom, wie beispielsweise einen String-Puffer oder eine Datei zu schreiben, die dann auf iteriert werden kann.

wie: Read a file in buffer from FTP python

jedoch beinhalten diese Lösungen die gesamte Datei in den Speicher geladen oder auf der Festplatte vor Beginn Herunterladen der Inhalte zu verarbeiten.

Ich habe nicht genug Speicher, um die gesamte Datei zu puffern und ich keinen Zugriff auf die Platte haben. Dies kann durch die Verarbeitung der Daten in der Callback-Funktion durchgeführt werden, aber Ich möchte wissen, ob es möglich ist, den FTP-Code in eine Magie zu wickeln, die einen Iterator zurückgibt, anstatt meinen Code mit Callbacks zu würzen.

I.E. statt:

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) 

Ich mag:

for chunk in get_ftp_data(): 
    do_stuff_to_chunk(chunk) 

und (im Gegensatz zu bestehenden Antworten) Ich möchte es tun, ohne vorher Iterieren auf die gesamte FTP-Datei auf der Festplatte oder Speicher zu schreiben.

Answer 1

Sie werden den retrbinary Anruf in einem anderen Thread setzen müssen und haben die Callback-Zuführblöcke zu einem Iterator:

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 

Wenn Sie keine Threads verwenden können, das Beste, was Sie tun können, Ihr Rückruf schreiben als Koroutine:

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