文件上传与静态资源

第 9 章 · 文件上传与静态资源

本章目标：使用 UploadFile 接收商品封面图；挂载 StaticFiles 提供上传文件与内置静态资源访问；实现图片格式、尺寸、大小校验；将文件路径写入 Product 模型；理解开发环境与生产 CDN（api.example.com / OSS）的差异。

学时建议：4～5 小时（含 2 小时上传跟练）

前置：完成 fastapi-web ch06～ch08（Product CRUD、JWT、统一异常）；ch05 ORM 模型。

9.1 静态资源 vs 用户上传

类型	来源	存储	访问方式
静态 Static	仓库内 CSS/默认图	`app/static/`	`StaticFiles` 或 Nginx
上传 Media	用户 POST multipart	`uploads/products/`	URL 如 `/media/products/xxx.webp`

开发环境 svc-demo
  POST /api/v1/products/{id}/cover  →  磁盘 uploads/
  GET  /media/products/abc.webp     →  StaticFiles

生产环境
  POST → 应用写本地或直传 OSS
  GET  → https://cdn.example.com/products/abc.webp

教学项目路径均为 svc-demo 本地目录；对外 URL 示例 https://api.example.com/media/...，不引用真实云账号。

9.2 模型扩展与迁移

app/models/product.py 增加字段：

cover_url: Mapped[str | None] = mapped_column(String(512), nullable=True)

alembic revision --autogenerate -m "add product cover_url"
alembic upgrade head

ProductRead Schema 同步：

class ProductRead(ProductBase):
    model_config = ConfigDict(from_attributes=True)
    id: int
    cover_url: str | None = None
    # ...

9.3 目录与配置

app/core/config.py：

from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent.parent.parent

class Settings(BaseSettings):
    # ...
    UPLOAD_DIR: Path = BASE_DIR / "uploads"
    MEDIA_URL: str = "/media"
    MAX_UPLOAD_SIZE: int = 5 * 1024 * 1024  # 5MB
    ALLOWED_IMAGE_TYPES: set[str] = {"image/jpeg", "image/png", "image/webp"}
    ALLOWED_IMAGE_EXT: set[str] = {".jpg", ".jpeg", ".png", ".webp"}

启动时确保目录存在（lifespan）：

@asynccontextmanager
async def lifespan(app: FastAPI):
    settings.UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
    (settings.UPLOAD_DIR / "products").mkdir(exist_ok=True)
    yield

.gitignore：

uploads/
!uploads/.gitkeep

9.4 挂载 StaticFiles

app/main.py：

from fastapi.staticfiles import StaticFiles

from app.core.config import settings

# 内置静态（可选：默认占位图）
app.mount("/static", StaticFiles(directory="app/static"), name="static")

# 用户上传媒体
app.mount(
    settings.MEDIA_URL,
    StaticFiles(directory=settings.UPLOAD_DIR),
    name="media",
)

挂载点	目录	用途
`/static`	`app/static`	内置资源
`/media`	`uploads/`	用户上传

访问示例：http://127.0.0.1:8000/media/products/550e8400.webp

注意：mount 应放在 路由注册之后 或注意路径不与 /api 冲突；/media 与 /api/v1 前缀不同，一般无冲突。

9.5 图片校验工具

app/utils/images.py：

import imghdr
import uuid
from pathlib import Path

from fastapi import UploadFile

from app.core.config import settings
from app.core.exceptions import AppException


async def read_limited(file: UploadFile, max_size: int) -> bytes:
    chunks: list[bytes] = []
    total = 0
    while True:
        chunk = await file.read(64 * 1024)
        if not chunk:
            break
        total += len(chunk)
        if total > max_size:
            raise AppException(
                code="FILE_TOO_LARGE",
                message=f"文件超过 {max_size // 1024 // 1024}MB 限制",
                status_code=413,
            )
        chunks.append(chunk)
    return b"".join(chunks)


def detect_image_type(data: bytes) -> str | None:
    kind = imghdr.what(None, h=data)
    mapping = {"jpeg": "image/jpeg", "png": "image/png", "webp": "image/webp"}
    return mapping.get(kind) if kind else None


def validate_image_content(data: bytes) -> str:
    mime = detect_image_type(data)
    if mime not in settings.ALLOWED_IMAGE_TYPES:
        raise AppException(
            code="INVALID_IMAGE",
            message="仅支持 JPEG、PNG、WebP",
            status_code=400,
        )
    return mime


def save_product_image(data: bytes, mime: str) -> str:
    ext_map = {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp"}
    ext = ext_map[mime]
    filename = f"{uuid.uuid4().hex}{ext}"
    dest = settings.UPLOAD_DIR / "products" / filename
    dest.write_bytes(data)
    return f"{settings.MEDIA_URL}/products/{filename}"

为何不只信 content_type：客户端可伪造；用魔数（imghdr 或 Pillow）校验真实格式。

Pillow 增强（可选）：