Python使用FastAPI实现文件上传接口的最佳方案

from fastapi import FastAPI, File, UploadFile
import shutil
import os

app = FastAPI()

UPLOAD_DIR = "uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    file_path = os.path.join(UPLOAD_DIR, file.filename)
    
    # 保存文件
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    return {"filename": file.filename, "content_type": file.content_type}

curl -F "file=@example.png" http://127.0.0.1:8000/upload/

{
  "filename": "example.png",
  "content_type": "image/png"
}

from typing import List
from fastapi import UploadFile, File

@app.post("/upload/multiple/")
async def upload_multiple(files: List[UploadFile] = File(...)):
    saved_files = []
    for file in files:
        file_path = os.path.join(UPLOAD_DIR, file.filename)
        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
        saved_files.append(file.filename)
    return {"uploaded_files": saved_files}

curl -F "files=@a.png" -F "files=@b.jpg" http://127.0.0.1:8000/upload/multiple/

import aiofiles

@app.post("/upload/async/")
async def upload_file_async(file: UploadFile = File(...)):
    file_path = os.path.join(UPLOAD_DIR, file.filename)
    async with aiofiles.open(file_path, 'wb') as out_file:
        while content := await file.read(1024):
            await out_file.write(content)
    await file.close()
    return {"filename": file.filename}

pip install aiofiles

client_max_body_size 10M;

ALLOWED_TYPES = ["image/png", "image/jpeg"]

@app.post("/upload/image/")
async def upload_image(file: UploadFile = File(...)):
    if file.content_type not in ALLOWED_TYPES:
        return {"error": "Unsupported file type"}
    # 继续保存逻辑...
    return {"filename": file.filename}

from fastapi import Form

@app.post("/upload/with_form/")
async def upload_with_form(
    user_id: int = Form(...),
    description: str = Form(""),
    file: UploadFile = File(...)
):
    file_path = os.path.join(UPLOAD_DIR, file.filename)
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    return {"user_id": user_id, "description": description, "filename": file.filename}