Как я могу использовать boto для потоковой передачи файла из Amazon S3 в Rackspace Cloudfiles?

Question

Я копирую файл с S3 в Cloudfiles, и я хотел бы избежать записи файла на диск. В библиотеке Python-Cloudfiles есть вызов object.stream(), который выглядит так, как мне нужно, но я не могу найти эквивалентный вызов в boto. Я надеюсь, что я смогу сделать что-то вроде:

shutil.copyfileobj(s3Object.stream(),rsObject.stream()) 

Возможно ли это с Бото (или я полагаю, любой другой s3 библиотеки)?

Answer 1

Ключевой объект в Бото, который представляет собой на объект в S3, можно использовать как итератор, так что вы должны быть в состоянии сделать что-то вроде этого:

>>> import boto 
>>> c = boto.connect_s3() 
>>> bucket = c.lookup('garnaat_pub') 
>>> key = bucket.lookup('Scan1.jpg') 
>>> for bytes in key: 
... write bytes to output stream 

Или, как в случае вашего примера , вы можете сделать:

>>> shutil.copyfileobj(key, rsObject.stream()) 

Answer 2

Я полагаю, что по крайней мере некоторые из людей, видящих этот вопрос будет, как я, и хотите способ потокового файл из Boto построчно (или запятую через запятую, или любой другой разделитель). Вот простой способ сделать это:

def getS3ResultsAsIterator(self, aws_access_info, key, prefix):   
    s3_conn = S3Connection(**aws_access) 
    bucket_obj = s3_conn.get_bucket(key) 
    # go through the list of files in the key 
    for f in bucket_obj.list(prefix=prefix): 
     unfinished_line = '' 
     for byte in f: 
      byte = unfinished_line + byte 
      #split on whatever, or use a regex with re.split() 
      lines = byte.split('\n') 
      unfinished_line = lines.pop() 
      for line in lines: 
       yield line 

@ Ответ garnaat, приведенный выше, по-прежнему велик и на 100% прав. Надеюсь, мой все еще помогает кому-то.

Answer 3

Другие ответы в этой теме относятся к boto, но S3.Object больше не итерируется в boto3. Таким образом, следующий не работает, он выдает сообщение об TypeError: 's3.Object' object is not iterable ошибки:

s3 = boto3.session.Session(profile_name=my_profile).resource('s3') 
    s3_obj = s3.Object(bucket_name=my_bucket, key=my_key) 

    with io.FileIO('sample.txt', 'w') as file: 
     for i in s3_obj: 
      file.write(i) 

В boto3 содержимом объекта доступно на S3.Object.get()['Body'], который не является итерацией либо, так что следующий еще не работает:

body = s3_obj.get()['Body'] 
    with io.FileIO('sample.txt', 'w') as file: 
     for i in body: 
      file.write(i) 

Таким образом, альтернативой является использование метода чтения, но это загружает весь объект S3 в памяти, которая при работе с большими файлами, не всегда есть возможность:

body = s3_obj.get()['Body'] 
    with io.FileIO('sample.txt', 'w') as file: 
     for i in body.read(): 
      file.write(i) 

Но метод read позволяет передать в параметре amt указание количества байтов, которые мы хотим прочитать из базового потока. Этот метод может быть неоднократно призывало, пока весь поток не считан:

body = s3_obj.get()['Body'] 
    with io.FileIO('sample.txt', 'w') as file: 
     while file.write(body.read(amt=512)): 
      pass 

Порывшись в botocore.response.StreamingBody код один понимает, что основной поток также доступен, таким образом, мы могли бы итерацию следующим образом:

body = s3_obj.get()['Body'] 
    with io.FileIO('sample.txt', 'w') as file: 
     for b in body._raw_stream: 
      file.write(b) 

В то время как прибегая к помощи я также видел некоторые ссылки, которые могут быть использовать, но я не пробовал:

• WrappedStreamingBody
• Another related thread
• An issue in boto3 github to request StreamingBody is a proper stream - который был закрыт !!!

Answer 4

Это мое решение упаковки потокового тела:

import io 
class S3ObjectInterator(io.RawIOBase): 
    def __init__(self, bucket, key): 
     """Initialize with S3 bucket and key names""" 
     self.s3c = boto3.client('s3') 
     self.obj_stream = self.s3c.get_object(Bucket=bucket, Key=key)['Body'] 

    def read(self, n=-1): 
     """Read from the stream""" 
     return self.obj_stream.read() if n == -1 else self.obj_stream.read(n) 

Пример использования:

obj_stream = S3ObjectInterator(bucket, key) 
for line in obj_stream: 
    print line