Django自定义storage上传文件到Minio

from io import BytesIO from django.core.files.storage import Storage from django.conf import settings from utils.minioClient import go_upload_file, go_delete_file, go_upload_file_have_size from utils.tools import img_bucket class MinioStorage(Storage): """ 自定义文件存储系统 """ def __init__(self): pass def _open(self, name, mode='rb'): """ 用于打开文件 :param name: 要打开的文件的名字 :param mode: 打开文件方式 :return: None """ pass def _save(self, name, content): """ 用于保存文件 :param name: 要保存的文件名字 :param content: 要保存的文件的内容 :return: None """ # print(name) name = name.replace('\\', '/') # ret = go_upload_file(BytesIO(content.read()), bucket=img_bucket, path_name=name) ret = go_upload_file_have_size(BytesIO(content.read()), content.size, bucket=img_bucket, path_name=name) assert ret == 1, '文件上传失败' path = '/' + name return path def delete(self, name): # im = get_file_path(instance.img) # print(name) # name = str(name).split('/')[-1] ret = go_delete_file(bucket=img_bucket, path_name=str(name)) # print(ret) def url(self, name): """ 返回name所指文件的绝对URL :param name: 要读取文件的引用 :return: """ host = settings.MINIOHTTP + settings.MINIOWEBHOST + ':' + settings.MINIOWEBPORT return host + "/" + img_bucket + name def exists(self, name): """ 检查文件存在 """ return False

import copy
import os

from django.conf import settings
from minio import Minio


MINIO_CONF = {
    'endpoint': settings.MINIOHOST + ':' + settings.MINIOPORT,
    'access_key': settings.MINIOUSER,
    'secret_key': settings.MINIOPWD,
    'secure': False
}

client = Minio(**MINIO_CONF)


def get_file_size(file):
    """
    获取文件大小
    file: bytes
    return: int
    """
    file.seek(0, os.SEEK_END)
    return file.tell()
    # im = io.BytesIO(file)
    # return im.getbuffer().nbytes


def go_upload_file(file, bucket='media', path_name=''):
    """
    上传文件
    """
    try:
        # print(path_name)
        # print(type(file))
        file_len = get_file_size(copy.copy(file))
        # print(file_len)
        client.put_object(bucket_name=bucket, object_name=path_name, data=file, length=file_len)
        return 1
    except Exception as e:
        print(e)
        return 0


def go_upload_file_have_size(file, size, bucket='media', path_name=''):
    """
    上传文件，已有文件大小
    """
    try:
        client.put_object(bucket_name=bucket, object_name=path_name, data=file, length=size)
        return 1
    except Exception as e:
        print(e)
        return 0


def go_delete_file(bucket='media', path_name=''):
    """
    删除文件
    """
    try:
        # print(bucket, path_name)
        client.remove_object(bucket_name=bucket, object_name=path_name)
        return 1
    except Exception as e:
        print(e)
        return 0


def go_delete_file_list(bucket='media', path_name_list=[]):
    """
    删除文件列表
    未实现，据说需要删除完遍历结果
    """
    try:
        ret = client.remove_objects(bucket, path_name_list, bypass_governance_mode=False)
        print(ret)
        return 1
    except Exception as e:
        print(e)
        return 0


def get_file_url(bucket='media', path_name=''):
    """
    获取文件url
    """
    try:
        url = client.presigned_get_object(bucket_name=bucket, object_name=path_name)
        return url
    except Exception as e:
        print(e)
        return None


def get_file_path(path):
    path = path.split('/')[2:]
    final_path = '/'.join(path)
    return final_path

import datetime import os import random import time img_type_list = ['.jpg', '.png', '.jpeg'] img_bucket = 'media' def get_secret(request): """ 获取加密的key """ return request.META.get('HTTP_AUTHORIZATION') or 'wchime' def get_time_string(): """ :return: 20220525140635467912 :PS ：并发较高时尾部随机数增加 """ time_string = str(datetime.datetime.fromtimestamp(time.time())).replace("-", "").replace(" ", "").replace(":","").replace(".", "") + str(random.randint(100, 999)) return time_string def split_file_type(file): """ 对文件名切割，获取名字和类型 """ file_li = os.path.splitext(file) return file_li[0], file_li[1] if __name__ == '__main__': im = 'a.png' s = split_file_type(im) print(s)

class TestImg(models.Model): name = models.CharField(verbose_name="名字", max_length=256) img = models.ImageField(upload_to='zzz')

from io import BytesIO
from django.core.files.base import ContentFile
from django.core.files.storage import default_storage
from django.utils.decorators import method_decorator
from rest_framework import status, serializers
from rest_framework.response import Response
from ani import models
from utils.decorators import request_decrypt
from utils.myView import MyPagination, MyView, MixinGetList, MixinPostCreateModel, MixinPutUpdateModel, \
MixinDeleteDestroyModel
from utils.tools import split_file_type, img_type_list, get_time_string, img_bucket
class TestImgSerializer(serializers.ModelSerializer):
class Meta:
model = models.TestImg
fields = '__all__'
def create(self, validated_data):
res = models.TestImg.objects.create(**validated_data)
return res
def update(self, instance, validated_data):
instance.name = validated_data.get('name')
if validated_data.get('img', False) is not False:
instance.img = validated_data.get('img')
instance.save()
return instance
@method_decorator(request_decrypt, name='get')
class TestImgView(MyView, MixinGetList):
queryset = models.TestImg.objects.all()
serializer_class = TestImgSerializer
all_serializer_class = TestImgSerializer
filter_class = ['name__icontains']
pagination_class = MyPagination
lookup_field = 'id'
ordeing_field = ('-id',)
def post(self, request, *args, **kwargs):
data = request.data
file_content = data.get('file').read()
file_name = get_time_string() + '.png'
# print(file_content)
content_file = ContentFile(file_content, file_name)
# print(content_file)
data['img'] = content_file
serializer_class = TestImgSerializer
serializer = serializer_class(data=data)
# print(serializer.is_valid())
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
def put(self, request, *args, **kwargs):
data = request.data
try:
instance = self.queryset.get(id=data.get(self.lookup_field))
except Exception:
return Response({'state': 'fail', 'msg': '未找到该数据'}, status=status.HTTP_400_BAD_REQUEST)
file = data.get('file')
if file:
file_content = file.read()
file_name = get_time_string() + '.png'
# print(file_content)
content_file = ContentFile(file_content, file_name)
# print(content_file)
data['img'] = content_file
if instance.img:
default_storage.delete(instance.img)
serializer_class = TestImgSerializer
serializer = serializer_class(instance=instance, data=data, partial=True)
# print(serializer.is_valid())
serializer.is_valid(raise_exception=True)
serializer.save()
return Response(serializer.data, status=status.HTTP_201_CREATED)
def delete(self, request, *args, **kwargs):
try:
instance = self.queryset.get(id=request.data.get(self.lookup_field))
except Exception:
return Response({'state': 'fail', 'msg': '未找到数据'}, status=status.HTTP_400_BAD_REQUEST)
default_storage.delete(instance.img)
instance.delete()
return Response({'state': 'success', 'msg': '删除成功'}, status=status.HTTP_204_NO_CONTENT)

DEFAULT_FILE_STORAGE = 'utils.MyStorage.MinioStorage'