DRF 文件上传实战 #

MEDIA_URL = '/media/'
MEDIA_ROOT = BASE_DIR / 'media'

INSTALLED_APPS = [
    ...
    'rest_framework',
]

from django.db import models
from django.contrib.auth.models import User
import os
import uuid

def upload_to(instance, filename):
    ext = filename.split('.')[-1]
    filename = f'{uuid.uuid4()}.{ext}'
    return os.path.join('uploads', str(instance.uploaded_at.year), str(instance.uploaded_at.month), filename)

class File(models.Model):
    FILE_TYPES = [
        ('image', '图片'),
        ('document', '文档'),
        ('video', '视频'),
        ('audio', '音频'),
        ('other', '其他'),
    ]
    
    file = models.FileField(upload_to=upload_to)
    filename = models.CharField(max_length=255)
    file_type = models.CharField(max_length=20, choices=FILE_TYPES)
    file_size = models.PositiveIntegerField()
    mime_type = models.CharField(max_length=100)
    uploader = models.ForeignKey(User, on_delete=models.CASCADE, related_name='files')
    description = models.TextField(blank=True)
    uploaded_at = models.DateTimeField(auto_now_add=True)
    
    class Meta:
        verbose_name = '文件'
        verbose_name_plural = '文件'
        ordering = ['-uploaded_at']
    
    def __str__(self):
        return self.filename
    
    def save(self, *args, **kwargs):
        if self.file:
            self.filename = self.file.name
            self.file_size = self.file.size
        super().save(*args, **kwargs)

from PIL import Image
from io import BytesIO
from django.core.files.base import ContentFile

class Image(models.Model):
    original = models.ImageField(upload_to='images/original/')
    thumbnail = models.ImageField(upload_to='images/thumbnail/', blank=True)
    medium = models.ImageField(upload_to='images/medium/', blank=True)
    uploader = models.ForeignKey(User, on_delete=models.CASCADE, related_name='images')
    alt_text = models.CharField(max_length=255, blank=True)
    uploaded_at = models.DateTimeField(auto_now_add=True)
    
    class Meta:
        verbose_name = '图片'
        verbose_name_plural = '图片'
        ordering = ['-uploaded_at']
    
    def save(self, *args, **kwargs):
        super().save(*args, **kwargs)
        
        if self.original and not self.thumbnail:
            self.create_thumbnails()
    
    def create_thumbnails(self):
        img = Image.open(self.original)
        
        # Thumbnail (150x150)
        thumb = img.copy()
        thumb.thumbnail((150, 150))
        thumb_io = BytesIO()
        thumb.save(thumb_io, format=img.format or 'JPEG')
        self.thumbnail.save(
            f'thumb_{self.original.name}',
            ContentFile(thumb_io.getvalue()),
            save=False
        )
        
        # Medium (800x800)
        medium = img.copy()
        medium.thumbnail((800, 800))
        medium_io = BytesIO()
        medium.save(medium_io, format=img.format or 'JPEG')
        self.medium.save(
            f'medium_{self.original.name}',
            ContentFile(medium_io.getvalue()),
            save=False
        )
        
        super().save()

class FileSerializer(serializers.ModelSerializer):
    file_url = serializers.SerializerMethodField()
    uploader_name = serializers.CharField(source='uploader.username', read_only=True)
    
    class Meta:
        model = File
        fields = ['id', 'file', 'file_url', 'filename', 'file_type', 'file_size', 
                  'mime_type', 'uploader', 'uploader_name', 'description', 'uploaded_at']
        read_only_fields = ['id', 'filename', 'file_type', 'file_size', 'mime_type', 'uploader', 'uploaded_at']
    
    def get_file_url(self, obj):
        request = self.context.get('request')
        if obj.file and request:
            return request.build_absolute_uri(obj.file.url)
        return None
    
    def create(self, validated_data):
        file = validated_data['file']
        
        validated_data['filename'] = file.name
        validated_data['file_size'] = file.size
        validated_data['mime_type'] = file.content_type
        
        import mimetypes
        mime_type = file.content_type
        if mime_type.startswith('image'):
            validated_data['file_type'] = 'image'
        elif mime_type.startswith('video'):
            validated_data['file_type'] = 'video'
        elif mime_type.startswith('audio'):
            validated_data['file_type'] = 'audio'
        elif 'pdf' in mime_type or 'document' in mime_type:
            validated_data['file_type'] = 'document'
        else:
            validated_data['file_type'] = 'other'
        
        return super().create(validated_data)

class ImageSerializer(serializers.ModelSerializer):
    original_url = serializers.SerializerMethodField()
    thumbnail_url = serializers.SerializerMethodField()
    medium_url = serializers.SerializerMethodField()
    
    class Meta:
        model = Image
        fields = ['id', 'original', 'original_url', 'thumbnail', 'thumbnail_url',
                  'medium', 'medium_url', 'alt_text', 'uploaded_at']
        read_only_fields = ['id', 'thumbnail', 'medium', 'uploaded_at']
    
    def get_original_url(self, obj):
        request = self.context.get('request')
        if obj.original and request:
            return request.build_absolute_uri(obj.original.url)
        return None
    
    def get_thumbnail_url(self, obj):
        request = self.context.get('request')
        if obj.thumbnail and request:
            return request.build_absolute_uri(obj.thumbnail.url)
        return None
    
    def get_medium_url(self, obj):
        request = self.context.get('request')
        if obj.medium and request:
            return request.build_absolute_uri(obj.medium.url)
        return None

class FileViewSet(viewsets.ModelViewSet):
    serializer_class = FileSerializer
    permission_classes = [IsAuthenticated]
    parser_classes = [MultiPartParser, FormParser]
    
    def get_queryset(self):
        queryset = File.objects.all()
        
        if not self.request.user.is_staff:
            queryset = queryset.filter(uploader=self.request.user)
        
        file_type = self.request.query_params.get('file_type')
        if file_type:
            queryset = queryset.filter(file_type=file_type)
        
        return queryset
    
    def perform_create(self, serializer):
        serializer.save(uploader=self.request.user)
    
    @action(detail=False, methods=['post'])
    def upload_multiple(self, request):
        files = request.FILES.getlist('files')
        
        if not files:
            return Response({'error': '请选择文件'}, status=400)
        
        results = []
        for file in files:
            file_obj = File.objects.create(
                file=file,
                filename=file.name,
                file_size=file.size,
                mime_type=file.content_type,
                uploader=request.user
            )
            results.append(FileSerializer(file_obj, context={'request': request}).data)
        
        return Response({'files': results}, status=201)

class ImageViewSet(viewsets.ModelViewSet):
    serializer_class = ImageSerializer
    permission_classes = [IsAuthenticated]
    parser_classes = [MultiPartParser, FormParser]
    
    def get_queryset(self):
        queryset = Image.objects.all()
        
        if not self.request.user.is_staff:
            queryset = queryset.filter(uploader=self.request.user)
        
        return queryset
    
    def perform_create(self, serializer):
        serializer.save(uploader=self.request.user)

from django.conf import settings
from django.conf.urls.static import static
from rest_framework.routers import DefaultRouter
from .views import FileViewSet, ImageViewSet

router = DefaultRouter()
router.register(r'files', FileViewSet)
router.register(r'images', ImageViewSet)

urlpatterns = [
    path('api/', include(router.urls)),
]

if settings.DEBUG:
    urlpatterns += static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)