Version 1 (modified by dobesv@…, 13 years ago) (diff)


Here's a class I created to integrate my site with amazon's storage service.

I like this because in the admin interface it looks pretty much like a file field and you can just click to upload. In theory you could integrate it into any oldforms system. Note that you have to set the amazon key, bucket, etc. manually, and that you have to put your amazon access keys into your site's settings module.

This hasn't been tested extensively, so I invite you to contribute any fixes right here on the page. Maybe the django gods will see fit to include this right into django.

from amazon import S3
from django.conf import settings
from django.db import models
from mimetypes import guess_type
from mx.DateTime import now
from django.core import validators
from django import oldforms
from django.dispatch import dispatcher
from django.utils.functional import curry
from django.utils.translation import gettext_lazy
from django.db.models import signals
import os

conn = S3.AWSAuthConnection(settings.AWS_ACCESS_KEY_ID,
generator = S3.QueryStringAuthGenerator(settings.AWS_ACCESS_KEY_ID,

class S3FileField(models.FileField):
    def __init__(self, verbose_name=None, name=None, bucket='', is_image=False, **kwargs):
        models.FileField.__init__(self, verbose_name, name, upload_to="s3", **kwargs)
        self.bucket = bucket
        self.is_image = is_image

    def get_manipulator_fields(self, opts, manipulator, change, name_prefix='', rel=False, follow=True):
        field_list = models.Field.get_manipulator_fields(self, opts, manipulator, change, name_prefix, rel, follow)
        if not self.blank:
            if rel:
                # This validator makes sure FileFields work in a related context.
                class RequiredFileField(object):
                    def __init__(self, other_field_names, other_file_field_name):
                        self.other_field_names = other_field_names
                        self.other_file_field_name = other_file_field_name
                        self.always_test = True
                    def __call__(self, field_data, all_data):
                        if not all_data.get(self.other_file_field_name, False):
                            c = validators.RequiredIfOtherFieldsGiven(self.other_field_names, gettext_lazy("This field is required."))
                            c(field_data, all_data)
                # First, get the core fields, if any.
                core_field_names = []
                for f in opts.fields:
                    if f.core and f != self:
                # Now, if there are any, add the validator to this FormField.
                if core_field_names:
                    field_list[0].validator_list.append(RequiredFileField(core_field_names, field_list[1].field_name))
                v = validators.RequiredIfOtherFieldNotGiven(field_list[1].field_name, gettext_lazy("This field is required."))
                v.always_test = True
                field_list[0].is_required = field_list[1].is_required = False

        return field_list
    def get_internal_type(self):
        return "FileField"
    def contribute_to_class(self, cls, name):
        super(S3FileField, self).contribute_to_class(cls, name)
        #models.CharField(maxlength=200, blank=self.blank, null=self.null).contribute_to_class(cls, "%s_filename"%(
        models.CharField(maxlength=200, blank=self.blank, null=self.null).contribute_to_class(cls, "%s_key"%(
        models.CharField(maxlength=200, blank=self.blank, null=self.null, default=settings.DEFAULT_BUCKET).contribute_to_class(cls, "%s_bucket"%(
        models.CharField(maxlength=200, blank=True, null=True, default="").contribute_to_class(cls, "%s_content_type"%(
        models.IntegerField(blank=True, null=True).contribute_to_class(cls, "%s_size"%(
        if self.is_image:
            models.IntegerField(blank=True, null=True).contribute_to_class(cls, "%s_width"%(
            models.IntegerField(blank=True, null=True).contribute_to_class(cls, "%s_height"%(
        # Getter for the file url
        def get_url(instance, field):
            return field.get_url(instance)
        setattr(cls, 'get_%s_url' %, curry(get_url, field=self))
        dispatcher.connect(self.delete_file, signal=signals.post_delete, sender=cls)

    def delete_file(self, instance):
        if getattr(instance, self.attname):
            bucket = self.get_bucket(instance)
            key = self.get_key(instance)
            conn.delete(bucket, key)
    def get_url(self, instance):
        print 'get_url called with instance %s'%instance
        bucket = self.get_bucket(instance)
        key = self.get_key(instance)
        print 'key %s bucket %s'%(key, bucket)
        import sys
        if bucket and key:
            url = generator.make_bare_url(bucket, key)
            print 'url', url
            return url
        return None
    def get_bucket(self, instance):
        return getattr(instance, "%s_bucket"
    def set_bucket(self, instance, bucket):    
        setattr(instance, "%s_bucket", bucket)
    def get_key(self, instance):
        return getattr(instance, "%s_key"
    def set_key(self, instance, key):    
        setattr(instance, "%s_key", key)
    def get_content_type(self, instance):
        return getattr(instance, "%s_content_type"
    def set_content_type(self, instance, content_type):    
        setattr(instance, "%s_content_type", content_type)
    def get_size(self, instance):
        return getattr(instance, "%s_size"
    def set_size(self, instance, size):    
        setattr(instance, "%s_size", size)
    def get_filename(self, instance):
        return getattr(instance,
    def set_filename(self, instance, filename):    
        setattr(instance,, filename)
    def get_width(self, instance):
        return getattr(instance, "%s_width"
    def set_width(self, instance, width):    
        setattr(instance, "%s_width", width)
    def get_height(self, instance):
        return getattr(instance, "%s_height"
    def set_height(self, instance, height):    
        setattr(instance, "%s_height", height)
    def get_manipulator_field_objs(self):
        if self.is_image: uploadType = oldforms.ImageUploadField
        else: uploadType = oldforms.FileUploadField
        return [uploadType, oldforms.HiddenField]

    def get_manipulator_field_names(self, name_prefix):
        return [name_prefix + + '_file', name_prefix +]

    def save_file(self, new_data, new_object, original_object, change, rel, save=True):
        upload_field_name = self.get_manipulator_field_names('')[0]
        if new_data.get(upload_field_name, False):
            if rel:
                new_content = new_data[upload_field_name][0]["content"]
                new_filename = new_data[upload_field_name][0]["filename"]
                new_content = new_data[upload_field_name]["content"]
                new_filename = new_data[upload_field_name]["filename"]
            self.set_filename(new_object, new_filename)
            key = new_data["%s_key"]
            bucket = new_data["%s_bucket"]
            if not bucket:
                bucket = self.bucket
            content_type = new_data["%s_content_type"]
            if new_filename and not content_type:
                content_type = guess_type(new_filename)[0]
                if not content_type:
                    content_type = "application/x-octet-stream"
            print 'Uploading %d bytes of data to bucket %s key %s filename %s'%(len(new_content), bucket, key, new_filename)
            conn.put(bucket, key, S3.S3Object(new_content), 
                     { 'x-amz-acl': 'public-read' , 'Content-Type': content_type})
            # TODO Calculate image width/height

Back to Top