从 fastapi UploadFile 将文件(不在 MEDIA_ROOT 中)保存到 Django FileField

问题描述 投票:0回答:1

我需要将 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'>

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

python django rest fastapi
1个回答
0
投票

因此,从 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
© www.soinside.com 2019 - 2024. All rights reserved.