bryanveloso / storage.py

Created May 28, 2009 — forked from robhudson/storage.py

from mimetypes import guess_type
from django.core.exceptions import ImproperlyConfigured
from django.core.files.storage import Storage
from django.utils.encoding import iri_to_uri
import re
 
try:
    import S3
except ImportError:
    raise ImproperlyConfigured, "Could not load amazon's S3 bindings.\
    \nSee http://developer.amazonwebservices.com/connect/entry.jspa?externalID=134"
 
try:    
    from settings import AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_STORAGE_BUCKET_NAME
except ImportError:
    raise ImproperlyConfigured, "AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_STORAGE_BUCKET_NAME required in settings.py."
 
try:
    from settings import AWS_CALLING_FORMAT
except ImportError:
    AWS_CALLING_FORMAT = S3.CallingFormat.PATH
 
class S3StorageException(Exception):
    """Exception raised when an S3 error occurs."""
 
class S3Storage(Storage):
    """
        Storage backend for Amazon Simple Storage Service.
    """
 
    def __init__(self, access_key=AWS_ACCESS_KEY_ID, secret_key=AWS_SECRET_ACCESS_KEY,
            bucket=AWS_STORAGE_BUCKET_NAME, acl='public-read', 
            calling_format=AWS_CALLING_FORMAT):
        self.bucket = bucket
        self.acl = acl
        self.connection = S3.AWSAuthConnection(access_key, secret_key, 
                            calling_format=calling_format)
        self.generator = S3.QueryStringAuthGenerator(access_key, secret_key, 
                            calling_format=calling_format, is_secure=False)
    
    def is_valid(self):
        """
        Tests if the connection is valid by checking the basic
        rest GET endpoint.
        """
        response = self.connection.list_all_my_buckets().http_response
        return response.status == 200
    
    def open(self, name, mode='rb', mixin=None):
        file = self.connection.get(self.bucket, name).object.data
        if mixin:
            # Add the mixin as a parent class of the File returned from storage.
            file.__class__ = type(mixin.__name__, (mixin, file.__class__), {})
        return file
    
    def save(self, name, content):
        # S3 name can be a path
        if name is None:
            name = content.name
        headers = {
            'x-amz-acl':  self.acl,
            'Content-Type': content.content_type or guess_type(name)[0] or "application/x-octet-stream",
            'Content-Disposition': 'filename="' + iri_to_uri(content.name) + '";',
        }
        response = self.connection.put(self.bucket, name, content.read(), headers)
        if response.http_response.status != 200:
            raise S3StorageException(response.message)
        return name
    
    def delete(self, name):
        self.connection.delete(self.bucket, name)
    
    def copy(self, old_name, name):
        try:
            filename = name.rsplit('/', 1)[1]
        except IndexError:
            filename = name
        # copy current content-type
        response = self.connection._make_request('HEAD', self.bucket, old_name)
        content_type = response.getheader('Content-Type')
        headers = {
            'x-amz-copy-source': '%s/%s' % (self.bucket, old_name),
            'x-amz-metadata-directive': 'REPLACE', # change content-disposition for new name
            'x-amz-acl':  self.acl, # otherwise would be reset to private
            'Content-Type': content_type,
            'Content-Disposition': 'filename="' + iri_to_uri(filename) + '";',
        }
        response = self.connection.copy(self.bucket, name, headers)
        if response.http_response.status != 200:
            raise S3StorageException(response.message)
        return name
    
    def move(self, old_name, name):
        # copy, then delete
        self.copy(old_name, name)
        self.delete(old_name)
 
    def exists(self, name):
        print '*** S3 EXISTS'
        #response = self.connection.make_request('HEAD', self.bucket, name)
        #print response
        #return response.status == 200
        raise NotImplementedError()
    
    def listbucket(self):
        response = self.connection.list_bucket(self.bucket)
        return [entry.key for entry in response.entries]
 
    def listdir(self, path):
        """
        Lists the contents of the specified path, returning a 2-tuple of lists;
        the first item being directories, the second item being files.
        """
        directories, files = [], []
        entries = self.listbucket()
 
        if path:
            path_re = re.compile('^(?P<path>%s/)(?P<subpath>.*)' % path)
        else: # root directory
            path_re = re.compile('^(?P<path>)(?P<subpath>.*)')
        
        for entry in entries:
            match = path_re.match(entry)
            if match:
                subpieces = match.group('subpath').split('/', 1)
                try: # check if this is a nested file
                    directory, more_path = subpieces
                    if not directory in directories:
                        directories.append(directory)
                except ValueError: # this is a file - no more path
                    files.append(subpieces[0])
 
        return directories, files
 
    def size(self, name):
        data = self.connection.get(self.bucket, name).object.data
        return len(data)
 
    def url(self, name):
        return self.generator.make_bare_url(self.bucket, name)