FastAPI 文件上传

# coding: utf8

from fastapi import FastAPI, File, UploadFile
from typing import List

app = FastAPI()

#  上传文单个件方法一，  File 是直接继承自 Form 的类。
@app.post('/file')
async def create(file: bytes = File(...)):
    return {"file_size": len(file)}


# 上传文单个件方法二
@app.post('/uploadfile')
async def upload_file(file: UploadFile = File(...)):
    # contents = await file.read()
    # print(contents)
    return {'filename': file.filename}

# 上传多个文件方法一
@app.post("/files")
async def create(fileS: List[bytes] = File(...)):
    return {"file_sizes": [len(file) for file in fileS]}


# 上传多个文件方法二
@app.post("/uploadfiles")
async def upload_file(fileS: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in fileS]}

# # 启动: uvicorn main:app --reload

pip install python-multipart

上传和form表单组合使用

- 上传单个多个文件
```python
from fastapi import FastAPI, File, Form, UploadFile
app = FastAPI()

@app.post("/files/")
async def create_file(
    file: bytes = File(...), one: UploadFile = File(...),
        token: str = Form(...)
):
    return {
        "filesize": len(file),
        "token": token,
        "one_content_type": one.content_type,
    }

• 上传多个文件

from fastapi import FastAPI, File, Form, UploadFile
from typing import List

app = FastAPI()


@app.post("/files/")
async def create_file(
    file: bytes = File(...), one: List[UploadFile] = File(...),
        token: str = Form(...)
):
    return {
        "filesize": len(file),
        "token": token,
        "one_content_type": [file.content_type for file in one],
    }

## UploadFile 的属性如下：

- filename：上传文件名字符串（str），例如， myimage.jpg；

- content_type：内容类型（MIME 类型 / 媒体类型）字符串（str），例如，image/jpeg；

- file： SpooledTemporaryFile（ file-like 对象）。其实就是 Python文件，可直接传递给其他预期 file-like 对象的函数或支持库。

- UploadFile 支持以下 async 方法，（使用内部 SpooledTemporaryFile）可调用相应的文件方法。

- write(data)：把 data （str 或 bytes）写入文件；

- read(size)：按指定数量的字节或字符（size (int)）读取文件内容；

- seek(offset)：移动至文件 offset （int）字节处的位置；

例如，await myfile.seek(0) 移动到文件开头；
执行 await myfile.read() 后，需再次读取已读取内容时，这种方法特别好用；
close()：关闭文件。

因为上述方法都是 async 方法，要搭配「await」使用。

例如，在 async 路径操作函数 内，要用以下方式读取文件内容

`contents = await myfile.read()`

文件上传保存

pip install python-multipart

import os
import shutil
from pathlib import Path
from typing import Union, Any
from tempfile import NamedTemporaryFile
from fastapi import APIRouter, Depends, File, UploadFile

from api.common.logger import logger
f
from api.utils import response_code

router = APIRouter()


@router.post("/upload/file/", summary="上传图片")
async def upload_image(
        file: UploadFile = File(...)
):
    logger.info(f"用户{token_data.sub}->上传文件:{file.filename}")

    # 本地存储临时方案，一般生产都是使用第三方云存储OSS(如七牛云, 阿里云)
    save_dir = f"{settings.BASE_DIR}/assets"
    if not os.path.exists(save_dir):
        os.mkdir(save_dir)

    try:
        suffix = Path(file.filename).suffix

        with NamedTemporaryFile(delete=False, suffix=suffix, dir=save_dir) as tmp:
            shutil.copyfileobj(file.file, tmp)
            tmp_file_name = Path(tmp.name).name
    finally:
        file.file.close()

    return response_code.resp_200(data={"image": f"http://127.0.0.1:8010/assets/{tmp_file_name}"})