import cStringIO as StringIO import os import logging import boto.s3.connection import boto.gs.connection import boto.s3.key import boto.gs.key from storage.basestorage import BaseStorage logger = logging.getLogger(__name__) class StreamReadKeyAsFile(object): def __init__(self, key): self._key = key self._finished = False def __enter__(self): return self def __exit__(self, type, value, tb): self._key.close(fast=True) def read(self, amt=None): if self._finished: return None resp = self._key.read(amt) if not resp: self._finished = True return resp class _CloudStorage(BaseStorage): def __init__(self, connection_class, key_class, upload_params, storage_path, access_key, secret_key, bucket_name): self._initialized = False self._bucket_name = bucket_name self._access_key = access_key self._secret_key = secret_key self._root_path = storage_path self._connection_class = connection_class self._key_class = key_class self._upload_params = upload_params self._cloud_conn = None self._cloud_bucket = None def _initialize_cloud_conn(self): if not self._initialized: self._cloud_conn = self._connection_class(self._access_key, self._secret_key) self._cloud_bucket = self._cloud_conn.get_bucket(self._bucket_name) self._initialized = True def _debug_key(self, key): """Used for debugging only.""" orig_meth = key.bucket.connection.make_request def new_meth(*args, **kwargs): print '#' * 16 print args print kwargs print '#' * 16 return orig_meth(*args, **kwargs) key.bucket.connection.make_request = new_meth def _init_path(self, path=None): path = os.path.join(self._root_path, path) if path else self._root_path if path and path[0] == '/': return path[1:] return path def get_content(self, path): self._initialize_cloud_conn() path = self._init_path(path) key = self._key_class(self._cloud_bucket, path) if not key.exists(): raise IOError('No such key: \'{0}\''.format(path)) return key.get_contents_as_string() def put_content(self, path, content): self._initialize_cloud_conn() path = self._init_path(path) key = self._key_class(self._cloud_bucket, path) key.set_contents_from_string(content, **self._upload_params) return path def get_supports_resumeable_downloads(self): return True def get_direct_download_url(self, path, expires_in=60): self._initialize_cloud_conn() path = self._init_path(path) k = self._key_class(self._cloud_bucket, path) return k.generate_url(expires_in) def stream_read(self, path): self._initialize_cloud_conn() path = self._init_path(path) key = self._key_class(self._cloud_bucket, path) if not key.exists(): raise IOError('No such key: \'{0}\''.format(path)) while True: buf = key.read(self.buffer_size) if not buf: break yield buf def stream_read_file(self, path): self._initialize_cloud_conn() path = self._init_path(path) key = self._key_class(self._cloud_bucket, path) if not key.exists(): raise IOError('No such key: \'{0}\''.format(path)) return StreamReadKeyAsFile(key) def stream_write(self, path, fp): # Minimum size of upload part size on S3 is 5MB self._initialize_cloud_conn() buffer_size = 5 * 1024 * 1024 if self.buffer_size > buffer_size: buffer_size = self.buffer_size path = self._init_path(path) mp = self._cloud_bucket.initiate_multipart_upload(path, **self._upload_params) num_part = 1 while True: try: buf = fp.read(buffer_size) if not buf: break io = StringIO.StringIO(buf) mp.upload_part_from_file(io, num_part) num_part += 1 io.close() except IOError: break mp.complete_upload() def list_directory(self, path=None): self._initialize_cloud_conn() path = self._init_path(path) if not path.endswith('/'): path += '/' ln = 0 if self._root_path != '/': ln = len(self._root_path) exists = False for key in self._cloud_bucket.list(prefix=path, delimiter='/'): exists = True name = key.name if name.endswith('/'): yield name[ln:-1] else: yield name[ln:] if exists is False: # In order to be compliant with the LocalStorage API. Even though # S3 does not have a concept of folders. raise OSError('No such directory: \'{0}\''.format(path)) def exists(self, path): self._initialize_cloud_conn() path = self._init_path(path) key = self._key_class(self._cloud_bucket, path) return key.exists() def remove(self, path): self._initialize_cloud_conn() path = self._init_path(path) key = self._key_class(self._cloud_bucket, path) if key.exists(): # It's a file key.delete() return # We assume it's a directory if not path.endswith('/'): path += '/' for key in self._cloud_bucket.list(prefix=path): key.delete() class S3Storage(_CloudStorage): def __init__(self, storage_path, s3_access_key, s3_secret_key, s3_bucket): upload_params = { 'encrypt_key': True, } super(S3Storage, self).__init__(boto.s3.connection.S3Connection, boto.s3.key.Key, upload_params, storage_path, s3_access_key, s3_secret_key, s3_bucket) class GoogleCloudStorage(_CloudStorage): def __init__(self, storage_path, access_key, secret_key, bucket_name): super(GoogleCloudStorage, self).__init__(boto.gs.connection.GSConnection, boto.gs.key.Key, {}, storage_path, access_key, secret_key, bucket_name) def stream_write(self, path, fp): # Minimum size of upload part size on S3 is 5MB self._initialize_cloud_conn() path = self._init_path(path) key = self._key_class(self._cloud_bucket, path) key.set_contents_from_stream(fp)