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
from core.config import settings
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}"})