从 fastapi UploadFile 将文件（不在 MEDIA_ROOT 中）保存到 Django FileField

Question

我需要将 fastapi

UploadFile

对象直接保存到下面定义的 Django 模型中：

class MyFile(models):
    file = models.FileField(upload_to="files/")

在 API 中，我设法使用

SpooledTemporaryFile

和

read

方法从

write

获取文件对象，如下块所示：

async def file_upload(file: UploadFile = File(...)):
    async with aiofiles.open(f"path/{file.filename}", "wb") as media:
        cont = await file.read()
        await media.write(cont)

然后我尝试将新文件转换为 Django 中的

FieldFile

类型，以便能够将其保存在

MyFile.file

字段中，但它不断返回

TypeError

并带有描述： write() 参数必须是 str，而不是生成器

from django.core.files import File as DjangoFile


async def file_upload(file: UploadFile = File(...)):
    ...

    async with aiofiles.open(f"path/{file.filename}") as media:
        d_file = DjangoFile(media)

        file_model = MyFile()
        file_model.file.save(file.filename, d_file)
        file_model.save()

    return {"ok": True}

堆栈跟踪显示

file_model.file.save(file.filename, d_file)

是有问题的行。

>>> type(file.filename)
str

>>> type(d_file)
<class 'django.core.files.base.File'>

你能指出我正确的方向吗？

Answer 1

因此，从 this 开始，我决定通过模型方法保存

MyFile

，该方法本质上检查

dir

中的最新文件，并在从

file

api 代码中调用时将其保存在

upload_file

字段中，如下所示：

async def file_upload(file: UploadFile = File(...)):
    async with aiofiles.open(f"path/{file.filename}", "wb") as media:
        cont = await file.read()
        await media.write(cont)
    
    # usage: implemented in MyFile.save
    d_file = MyFile()
    await d_file.asave()

然后在模型中我重构了 save() 方法，如下所示：

class MyFile(models):
    file = models.FileField(upload_to="files/")

    def save(self, *args, **kwargs):
        filename = self.generate_file()

        with open(filename, "rb") as f:
            self.file.save(filename, f, save=False)  # False to avoid recursion error
        return super(File, self).save(*args, **kwargs)

    async def asave(
        self, force_insert=False, force_update=False, using=None, update_fields=None
    ):
        return await sync_to_async(self.save)(
            force_insert=force_insert,
            force_update=force_update,
            using=using,
            update_fields=update_fields,
        )

    @staticmethod
    def generate_file():
        import glob
        import os

        files = glob.glob("path/*")  # * means all, if specific format needed then *.csv
        latest = max(files, key=os.path.getctime)
        return latest