<s> import os <EOL> import posixpath <EOL> import mimetypes <EOL> from datetime import datetime <EOL> from gzip import GzipFile <EOL> from tempfile import SpooledTemporaryFile <EOL> from django . core . files . base import File <EOL> from django . core . exceptions import ImproperlyConfigured , SuspiciousOperation <EOL> from django . utils . encoding import force_text , smart_str , filepath_to_uri , force_bytes <EOL> try : <EOL> from boto import __version__ as boto_version <EOL> from boto . s3 . connection import S3Connection , SubdomainCallingFormat <EOL> from boto . exception import S3ResponseError <EOL> from boto . s3 . key import Key as S3Key <EOL> from boto . utils import parse_ts , ISO8601 <EOL> except ImportError : <EOL> raise ImproperlyConfigured ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> from storages . utils import setting <EOL> from storages . compat import urlparse , BytesIO , deconstructible , Storage <EOL> boto_version_info = tuple ( [ int ( i ) for i in boto_version . split ( '<STR_LIT:->' ) [ <NUM_LIT:0> ] . split ( '<STR_LIT:.>' ) ] ) <EOL> if boto_version_info [ : <NUM_LIT:2> ] < ( <NUM_LIT:2> , <NUM_LIT:32> ) : <EOL> raise ImproperlyConfigured ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def safe_join ( base , * paths ) : <EOL> """<STR_LIT>""" <EOL> base_path = force_text ( base ) <EOL> base_path = base_path . rstrip ( '<STR_LIT:/>' ) <EOL> paths = [ force_text ( p ) for p in paths ] <EOL> final_path = base_path <EOL> for path in paths : <EOL> final_path = urlparse . urljoin ( final_path . rstrip ( '<STR_LIT:/>' ) + "<STR_LIT:/>" , path ) <EOL> base_path_len = len ( base_path ) <EOL> if ( not final_path . startswith ( base_path ) or <EOL> final_path [ base_path_len : base_path_len + <NUM_LIT:1> ] not in ( '<STR_LIT>' , '<STR_LIT:/>' ) ) : <EOL> raise ValueError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return final_path . lstrip ( '<STR_LIT:/>' ) <EOL> @ deconstructible <EOL> class S3BotoStorageFile ( File ) : <EOL> """<STR_LIT>""" <EOL> buffer_size = setting ( '<STR_LIT>' , <NUM_LIT> ) <EOL> def __init__ ( self , name , mode , storage , buffer_size = None ) : <EOL> self . _storage = storage <EOL> self . name = name [ len ( self . _storage . location ) : ] . lstrip ( '<STR_LIT:/>' ) <EOL> self . _mode = mode <EOL> self . key = storage . bucket . get_key ( self . _storage . _encode_name ( name ) ) <EOL> if not self . key and '<STR_LIT:w>' in mode : <EOL> self . key = storage . bucket . new_key ( storage . _encode_name ( name ) ) <EOL> self . _is_dirty = False <EOL> self . _file = None <EOL> self . _multipart = None <EOL> if buffer_size is not None : <EOL> self . buffer_size = buffer_size <EOL> self . _write_counter = <NUM_LIT:0> <EOL> @ property <EOL> def size ( self ) : <EOL> return self . key . size <EOL> def _get_file ( self ) : <EOL> if self . _file is None : <EOL> self . _file = SpooledTemporaryFile ( <EOL> max_size = self . _storage . max_memory_size , <EOL> suffix = "<STR_LIT>" , <EOL> dir = setting ( "<STR_LIT>" , None ) <EOL> ) <EOL> if '<STR_LIT:r>' in self . _mode : <EOL> self . _is_dirty = False <EOL> self . key . get_contents_to_file ( self . _file ) <EOL> self . _file . seek ( <NUM_LIT:0> ) <EOL> if self . _storage . gzip and self . key . content_encoding == '<STR_LIT>' : <EOL> self . _file = GzipFile ( mode = self . _mode , fileobj = self . _file ) <EOL> return self . _file <EOL> def _set_file ( self , value ) : <EOL> self . _file = value <EOL> file = property ( _get_file , _set_file ) <EOL> def read ( self , * args , ** kwargs ) : <EOL> if '<STR_LIT:r>' not in self . _mode : <EOL> raise AttributeError ( "<STR_LIT>" ) <EOL> return super ( S3BotoStorageFile , self ) . read ( * args , ** kwargs ) <EOL> def write ( self , content , * args , ** kwargs ) : <EOL> if '<STR_LIT:w>' not in self . _mode : <EOL> raise AttributeError ( "<STR_LIT>" ) <EOL> self . _is_dirty = True <EOL> if self . _multipart is None : <EOL> provider = self . key . bucket . connection . provider <EOL> upload_headers = { <EOL> provider . acl_header : self . _storage . default_acl <EOL> } <EOL> upload_headers . update ( { '<STR_LIT:Content-Type>' : mimetypes . guess_type ( self . key . name ) [ <NUM_LIT:0> ] or self . _storage . key_class . DefaultContentType } ) <EOL> upload_headers . update ( self . _storage . headers ) <EOL> self . _multipart = self . _storage . bucket . initiate_multipart_upload ( <EOL> self . key . name , <EOL> headers = upload_headers , <EOL> reduced_redundancy = self . _storage . reduced_redundancy , <EOL> encrypt_key = self . _storage . encryption , <EOL> ) <EOL> if self . buffer_size <= self . _buffer_file_size : <EOL> self . _flush_write_buffer ( ) <EOL> return super ( S3BotoStorageFile , self ) . write ( force_bytes ( content ) , * args , ** kwargs ) <EOL> @ property <EOL> def _buffer_file_size ( self ) : <EOL> pos = self . file . tell ( ) <EOL> self . file . seek ( <NUM_LIT:0> , os . SEEK_END ) <EOL> length = self . file . tell ( ) <EOL> self . file . seek ( pos ) <EOL> return length <EOL> def _flush_write_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _buffer_file_size : <EOL> self . _write_counter += <NUM_LIT:1> <EOL> self . file . seek ( <NUM_LIT:0> ) <EOL> headers = self . _storage . headers . copy ( ) <EOL> self . _multipart . upload_part_from_file ( <EOL> self . file , self . _write_counter , headers = headers ) <EOL> def close ( self ) : <EOL> if self . _is_dirty : <EOL> self . _flush_write_buffer ( ) <EOL> self . _multipart . complete_upload ( ) <EOL> else : <EOL> if not self . _multipart is None : <EOL> self . _multipart . cancel_upload ( ) <EOL> self . key . close ( ) <EOL> if self . _file is not None : <EOL> self . _file . close ( ) <EOL> self . _file = None <EOL> @ deconstructible <EOL> class S3BotoStorage ( Storage ) : <EOL> """<STR_LIT>""" <EOL> connection_class = S3Connection <EOL> connection_response_error = S3ResponseError <EOL> file_class = S3BotoStorageFile <EOL> key_class = S3Key <EOL> access_key_names = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> secret_key_names = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> access_key = setting ( '<STR_LIT>' , setting ( '<STR_LIT>' ) ) <EOL> secret_key = setting ( '<STR_LIT>' , setting ( '<STR_LIT>' ) ) <EOL> file_overwrite = setting ( '<STR_LIT>' , True ) <EOL> headers = setting ( '<STR_LIT>' , { } ) <EOL> bucket_name = setting ( '<STR_LIT>' ) <EOL> auto_create_bucket = setting ( '<STR_LIT>' , False ) <EOL> default_acl = setting ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> bucket_acl = setting ( '<STR_LIT>' , default_acl ) <EOL> querystring_auth = setting ( '<STR_LIT>' , True ) <EOL> querystring_expire = setting ( '<STR_LIT>' , <NUM_LIT> ) <EOL> reduced_redundancy = setting ( '<STR_LIT>' , False ) <EOL> location = setting ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> encryption = setting ( '<STR_LIT>' , False ) <EOL> custom_domain = setting ( '<STR_LIT>' ) <EOL> calling_format = setting ( '<STR_LIT>' , SubdomainCallingFormat ( ) ) <EOL> secure_urls = setting ( '<STR_LIT>' , True ) <EOL> file_name_charset = setting ( '<STR_LIT>' , '<STR_LIT:utf-8>' ) <EOL> gzip = setting ( '<STR_LIT>' , False ) <EOL> preload_metadata = setting ( '<STR_LIT>' , False ) <EOL> gzip_content_types = setting ( '<STR_LIT>' , ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) ) <EOL> url_protocol = setting ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> host = setting ( '<STR_LIT>' , S3Connection . DefaultHost ) <EOL> use_ssl = setting ( '<STR_LIT>' , True ) <EOL> port = setting ( '<STR_LIT>' , None ) <EOL> proxy = setting ( '<STR_LIT>' , None ) <EOL> proxy_port = setting ( '<STR_LIT>' , None ) <EOL> max_memory_size = setting ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> def __init__ ( self , acl = None , bucket = None , ** settings ) : <EOL> for name , value in settings . items ( ) : <EOL> if hasattr ( self , name ) : <EOL> setattr ( self , name , value ) <EOL> if acl is not None : <EOL> self . default_acl = acl <EOL> if bucket is not None : <EOL> self . bucket_name = bucket <EOL> self . location = ( self . location or '<STR_LIT>' ) . lstrip ( '<STR_LIT:/>' ) <EOL> if self . secure_urls : <EOL> self . url_protocol = '<STR_LIT>' <EOL> self . _entries = { } <EOL> self . _bucket = None <EOL> self . _connection = None <EOL> if not self . access_key and not self . secret_key : <EOL> self . access_key , self . secret_key = self . _get_access_keys ( ) <EOL> @ property <EOL> def connection ( self ) : <EOL> if self . _connection is None : <EOL> self . _connection = self . connection_class ( <EOL> self . access_key , <EOL> self . secret_key , <EOL> is_secure = self . use_ssl , <EOL> calling_format = self . calling_format , <EOL> host = self . host , <EOL> port = self . port , <EOL> proxy = self . proxy , <EOL> proxy_port = self . proxy_port <EOL> ) <EOL> return self . _connection <EOL> @ property <EOL> def bucket ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _bucket is None : <EOL> self . _bucket = self . _get_or_create_bucket ( self . bucket_name ) <EOL> return self . _bucket <EOL> @ property <EOL> def entries ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . preload_metadata and not self . _entries : <EOL> self . _entries = dict ( ( self . _decode_name ( entry . key ) , entry ) <EOL> for entry in self . bucket . list ( prefix = self . location ) ) <EOL> return self . _entries <EOL> def _get_access_keys ( self ) : <EOL> """<STR_LIT>""" <EOL> def lookup_env ( names ) : <EOL> for name in names : <EOL> value = os . environ . get ( name ) <EOL> if value : <EOL> return value <EOL> access_key = self . access_key or lookup_env ( self . access_key_names ) <EOL> secret_key = self . secret_key or lookup_env ( self . secret_key_names ) <EOL> return access_key , secret_key <EOL> def _get_or_create_bucket ( self , name ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . connection . get_bucket ( name , validate = self . auto_create_bucket ) <EOL> except self . connection_response_error : <EOL> if self . auto_create_bucket : <EOL> bucket = self . connection . create_bucket ( name ) <EOL> bucket . set_acl ( self . bucket_acl ) <EOL> return bucket <EOL> raise ImproperlyConfigured ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % name ) <EOL> def _clean_name ( self , name ) : <EOL> """<STR_LIT>""" <EOL> clean_name = posixpath . normpath ( name ) . replace ( '<STR_LIT:\\>' , '<STR_LIT:/>' ) <EOL> if name . endswith ( '<STR_LIT:/>' ) and not clean_name . endswith ( '<STR_LIT:/>' ) : <EOL> return clean_name + '<STR_LIT:/>' <EOL> else : <EOL> return clean_name <EOL> def _normalize_name ( self , name ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return safe_join ( self . location , name ) <EOL> except ValueError : <EOL> raise SuspiciousOperation ( "<STR_LIT>" % <EOL> name ) <EOL> def _encode_name ( self , name ) : <EOL> return smart_str ( name , encoding = self . file_name_charset ) <EOL> def _decode_name ( self , name ) : <EOL> return force_text ( name , encoding = self . file_name_charset ) <EOL> def _compress_content ( self , content ) : <EOL> """<STR_LIT>""" <EOL> zbuf = BytesIO ( ) <EOL> zfile = GzipFile ( mode = '<STR_LIT:wb>' , compresslevel = <NUM_LIT:6> , fileobj = zbuf ) <EOL> try : <EOL> zfile . write ( force_bytes ( content . read ( ) ) ) <EOL> finally : <EOL> zfile . close ( ) <EOL> zbuf . seek ( <NUM_LIT:0> ) <EOL> content . file = zbuf <EOL> content . seek ( <NUM_LIT:0> ) <EOL> return content <EOL> def _open ( self , name , mode = '<STR_LIT:rb>' ) : <EOL> name = self . _normalize_name ( self . _clean_name ( name ) ) <EOL> f = self . file_class ( name , mode , self ) <EOL> if not f . key : <EOL> raise IOError ( '<STR_LIT>' % name ) <EOL> return f <EOL> def _save ( self , name , content ) : <EOL> cleaned_name = self . _clean_name ( name ) <EOL> name = self . _normalize_name ( cleaned_name ) <EOL> headers = self . headers . copy ( ) <EOL> _type , encoding = mimetypes . guess_type ( name ) <EOL> content_type = getattr ( content , '<STR_LIT>' , <EOL> _type or self . key_class . DefaultContentType ) <EOL> headers . update ( { '<STR_LIT:Content-Type>' : content_type } ) <EOL> if self . gzip and content_type in self . gzip_content_types : <EOL> content = self . _compress_content ( content ) <EOL> headers . update ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> elif encoding : <EOL> headers . update ( { '<STR_LIT>' : encoding } ) <EOL> content . name = cleaned_name <EOL> encoded_name = self . _encode_name ( name ) <EOL> key = self . bucket . get_key ( encoded_name ) <EOL> if not key : <EOL> key = self . bucket . new_key ( encoded_name ) <EOL> if self . preload_metadata : <EOL> self . _entries [ encoded_name ] = key <EOL> key . last_modified = datetime . utcnow ( ) . strftime ( ISO8601 ) <EOL> key . set_metadata ( '<STR_LIT:Content-Type>' , content_type ) <EOL> self . _save_content ( key , content , headers = headers ) <EOL> return cleaned_name <EOL> def _save_content ( self , key , content , headers ) : <EOL> kwargs = { } <EOL> if self . encryption : <EOL> kwargs [ '<STR_LIT>' ] = self . encryption <EOL> key . set_contents_from_file ( content , headers = headers , <EOL> policy = self . default_acl , <EOL> reduced_redundancy = self . reduced_redundancy , <EOL> rewind = True , ** kwargs ) <EOL> def delete ( self , name ) : <EOL> name = self . _normalize_name ( self . _clean_name ( name ) ) <EOL> self . bucket . delete_key ( self . _encode_name ( name ) ) <EOL> def exists ( self , name ) : <EOL> if not name : <EOL> try : <EOL> self . bucket <EOL> return True <EOL> except ImproperlyConfigured : <EOL> return False <EOL> name = self . _normalize_name ( self . _clean_name ( name ) ) <EOL> if self . entries : <EOL> return name in self . entries <EOL> k = self . bucket . new_key ( self . _encode_name ( name ) ) <EOL> return k . exists ( ) <EOL> def listdir ( self , name ) : <EOL> name = self . _normalize_name ( self . _clean_name ( name ) ) <EOL> if name and not name . endswith ( '<STR_LIT:/>' ) : <EOL> name += '<STR_LIT:/>' <EOL> dirlist = self . bucket . list ( self . _encode_name ( name ) ) <EOL> files = [ ] <EOL> dirs = set ( ) <EOL> base_parts = name . split ( "<STR_LIT:/>" ) [ : - <NUM_LIT:1> ] <EOL> for item in dirlist : <EOL> parts = item . name . split ( "<STR_LIT:/>" ) <EOL> parts = parts [ len ( base_parts ) : ] <EOL> if len ( parts ) == <NUM_LIT:1> : <EOL> files . append ( parts [ <NUM_LIT:0> ] ) <EOL> elif len ( parts ) > <NUM_LIT:1> : <EOL> dirs . add ( parts [ <NUM_LIT:0> ] ) <EOL> return list ( dirs ) , files <EOL> def size ( self , name ) : <EOL> name = self . _normalize_name ( self . _clean_name ( name ) ) <EOL> if self . entries : <EOL> entry = self . entries . get ( name ) <EOL> if entry : <EOL> return entry . size <EOL> return <NUM_LIT:0> <EOL> return self . bucket . get_key ( self . _encode_name ( name ) ) . size <EOL> def modified_time ( self , name ) : <EOL> name = self . _normalize_name ( self . _clean_name ( name ) ) <EOL> entry = self . entries . get ( name ) <EOL> if entry is None : <EOL> entry = self . bucket . get_key ( self . _encode_name ( name ) ) <EOL> return parse_ts ( entry . last_modified ) <EOL> def url ( self , name , headers = None , response_headers = None , expire = None ) : <EOL> name = self . _normalize_name ( self . _clean_name ( name ) ) <EOL> if self . custom_domain : <EOL> return "<STR_LIT>" % ( self . url_protocol , <EOL> self . custom_domain , filepath_to_uri ( name ) ) <EOL> if expire is None : <EOL> expire = self . querystring_expire <EOL> return self . connection . generate_url ( <EOL> expire , <EOL> method = '<STR_LIT:GET>' , <EOL> bucket = self . bucket . name , <EOL> key = self . _encode_name ( name ) , <EOL> headers = headers , <EOL> query_auth = self . querystring_auth , <EOL> force_http = not self . secure_urls , <EOL> response_headers = response_headers , <EOL> ) <EOL> def get_available_name ( self , name , max_length = None ) : <EOL> """<STR_LIT>""" <EOL> if self . file_overwrite : <EOL> name = self . _clean_name ( name ) <EOL> return name <EOL> return super ( S3BotoStorage , self ) . get_available_name ( name , max_length ) </s>
