文件上传

from fastapi import FastAPI, File, UploadFile
 
app = FastAPI()
 
@app.post("/upload/avatar")
async def upload_avatar(file: UploadFile = File(...)):
    return {
        "filename": file.filename,
        "content_type": file.content_type,
    }

from typing import List
from fastapi import FastAPI, File, UploadFile
 
app = FastAPI()
 
@app.post("/upload/resources")
async def upload_resources(files: List[UploadFile] = File(...)):
    return [
        {
            "filename": f.filename,
            "content_type": f.content_type,
        }
        for f in files
    ]

from fastapi import Form
 
@app.post("/upload/lesson-assets")
async def upload_lesson_assets(
    lesson_id: int = Form(...),
    description: str = Form(""),
    files: List[UploadFile] = File(...)
):
    return {
        "lesson_id": lesson_id,
        "description": description,
        "files": [
            {"filename": f.filename, "content_type": f.content_type}
            for f in files
        ],
    }

from pathlib import Path
from fastapi import FastAPI, File, UploadFile, HTTPException
 
app = FastAPI()
MAX_FILE_SIZE = 5 * 1024 * 1024  # 5MB
 
@app.post("/upload/image")
async def upload_image(file: UploadFile = File(...)):
    contents = await file.read()
    if len(contents) > MAX_FILE_SIZE:
        raise HTTPException(status_code=400, detail="文件过大，最大允许 5MB")
    await file.seek(0)
    return {"filename": file.filename}

ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png"}
ALLOWED_MIME_PREFIXES = {"image/"}
 
def validate_file_type(upload_file: UploadFile) -> None:
    suffix = Path(upload_file.filename).suffix.lower()
    if suffix not in ALLOWED_EXTENSIONS:
        raise HTTPException(status_code=400, detail="不支持的文件扩展名")
    if not any(upload_file.content_type.startswith(p) for p in ALLOWED_MIME_PREFIXES):
        raise HTTPException(status_code=400, detail="不支持的内容类型")

import uuid
import shutil
from pathlib import Path
from fastapi import FastAPI, UploadFile, File
 
app = FastAPI()
UPLOAD_ROOT = Path("uploads")
UPLOAD_ROOT.mkdir(exist_ok=True)
 
async def save_file_for_course(course_id: int, file: UploadFile) -> str:
    course_dir = UPLOAD_ROOT / f"course_{course_id}"
    course_dir.mkdir(exist_ok=True)
    unique_name = f"{uuid.uuid4().hex}{Path(file.filename).suffix.lower()}"
    destination = course_dir / unique_name
    with destination.open("wb") as out_file:
        shutil.copyfileobj(file.file, out_file)
    return str(destination)
 
@app.post("/courses/{course_id}/cover")
async def upload_course_cover(course_id: int, file: UploadFile = File(...)):
    validate_file_type(file)
    path = await save_file_for_course(course_id, file)
    return {"path": path}

async def save_large_file(destination: Path, upload_file: UploadFile) -> None:
    with destination.open("wb") as out_file:
        while True:
            chunk = await upload_file.read(1024 * 1024)
            if not chunk:
                break
            out_file.write(chunk)
    await upload_file.seek(0)

from fastapi import BackgroundTasks
 
def process_file_async(path: str) -> None:
    # 这里可以是视频转码、生成缩略图、文本提取等
    ...
 
@app.post("/upload/video")
async def upload_video(
    background_tasks: BackgroundTasks,
    file: UploadFile = File(...)
):
    validate_file_type(file)
    dest = UPLOAD_ROOT / f"video_{uuid.uuid4().hex}{Path(file.filename).suffix.lower()}"
    await save_large_file(dest, file)
    background_tasks.add_task(process_file_async, str(dest))
    return {"message": "视频已接收，处理将在后台继续进行"}

from fastapi import Depends
 
@app.post("/me/avatar")
async def upload_my_avatar(
    file: UploadFile = File(...),
    current_user: User = Depends(get_current_active_user),
):
    validate_file_type(file)
    user_dir = UPLOAD_ROOT / f"user_{current_user.username}"
    user_dir.mkdir(exist_ok=True)
    dest = user_dir / f"avatar{Path(file.filename).suffix.lower()}"
    await save_large_file(dest, file)
    return {"message": "头像已上传", "user": current_user.username}
 
@app.post("/admin/import/users")
async def upload_user_import_file(
    file: UploadFile = File(...),
    admin_user: User = Depends(require_role("admin")),
):
    validate_file_type(file)
    dest = UPLOAD_ROOT / f"import_{uuid.uuid4().hex}{Path(file.filename).suffix.lower()}"
    await save_large_file(dest, file)
    return {"message": "导入文件已接收", "uploaded_by": admin_user.username}

from typing import Protocol
 
class StorageService(Protocol):
    async def save(self, key: str, file: UploadFile) -> str:
        ...
 
class LocalStorageService:
    def __init__(self, root: Path):
        self.root = root
 
    async def save(self, key: str, file: UploadFile) -> str:
        dest = self.root / key
        dest.parent.mkdir(parents=True, exist_ok=True)
        await save_large_file(dest, file)
        return str(dest)
 
storage_service = LocalStorageService(UPLOAD_ROOT)
 
async def get_storage() -> StorageService:
    return storage_service
 
@app.post("/upload/course/{course_id}/material")
async def upload_course_material(
    course_id: int,
    file: UploadFile = File(...),
    storage: StorageService = Depends(get_storage),
):
    validate_file_type(file)
    key = f"courses/{course_id}/{uuid.uuid4().hex}{Path(file.filename).suffix.lower()}"
    path = await storage.save(key, file)
    return {"key": key, "path": path}