12. FastAPI 文件上传

本章分为三部分：单文件上传、多文件上传、上传到阿里云 OSS。示例同时支持常见图片和普通文档，并演示大小限制、类型白名单、真实内容检查、唯一存储名、防止覆盖以及失败清理。

示例允许的类型为 JPG、JPEG、PNG 和 PDF。不要盲目允许所有扩展名，尤其不要直接开放可执行文件或 HTML、SVG 等主动内容。

一、单文件上传（图片或者文件）

1. 本例的上传规则

规则	本例约定	不满足时的响应
单文件大小	最大 10 MiB，按实际读取字节数计算	413
空文件	不允许	400
扩展名	只允许 `.jpg`、`.jpeg`、`.png`、`.pdf`	415
声明类型	`content_type` 必须与扩展名白名单匹配	415
真实内容	图片使用 Pillow 验证，PDF 检查文件头	415
存储名	使用 `UUID + 扩展名`，不直接使用用户文件名	避免重名与路径攻击
防止覆盖	以独占模式 `xb` 创建文件	不覆盖已有文件
失败处理	校验或写入失败时删除不完整文件	不保留垃圾文件

这里的“不能重名”是指服务器存储名不能重复，旧文件不能被覆盖。两个用户都上传名为 avatar.png 的文件时，会得到不同的存储名；原始文件名只作为元数据返回。

先安装依赖：

powershell

uv add python-multipart pillow

2. 后端完整代码

保存为 main.py：

python

from pathlib import Path
from typing import Annotated
from uuid import uuid4
import uvicorn
from fastapi import FastAPI, File, HTTPException, UploadFile
from PIL import Image, UnidentifiedImageError
from fastapi.staticfiles import StaticFiles
app = FastAPI(title="单文件与多文件上传示例")

BASE_DIR = Path(__file__).resolve().parent
UPLOAD_DIR = BASE_DIR / "uploads"
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)

MAX_FILE_SIZE = 10 * 1024 * 1024
CHUNK_SIZE = 1024 * 1024

ALLOWED_TYPES = {
    ".jpg": {"image/jpeg"},
    ".jpeg": {"image/jpeg"},
    ".png": {"image/png"},
    ".pdf": {"application/pdf"},
}


def validate_metadata(file: UploadFile) -> str:
    """校验上传文件的名称、扩展名和客户端声明的 MIME 类型。

    参数:
        file: FastAPI 接收到的上传文件。
    返回:
        通过校验的小写文件扩展名，例如 ".jpg"。
    异常:
        HTTPException: 扩展名不受支持，或 MIME 类型与扩展名不匹配。
    说明:
        MIME 类型由客户端提供，因此这里只做初步检查，还需验证真实文件内容。
    """
    original_name = Path(file.filename or "").name
    suffix = Path(original_name).suffix.lower()

    if suffix not in ALLOWED_TYPES:
        raise HTTPException(
            status_code=415,
            detail="只支持 JPG、JPEG、PNG 和 PDF 文件",
        )

    if file.content_type not in ALLOWED_TYPES[suffix]:
        raise HTTPException(
            status_code=415,
            detail="文件扩展名与 content_type 不匹配",
        )

    return suffix


def validate_content(path: Path, suffix: str) -> None:
    """检查已保存文件的真实内容，防止伪造扩展名或 MIME 类型。

    PDF 通过文件头进行初步验证；JPG、JPEG 和 PNG 使用 Pillow 核对真实
    格式并检查文件完整性。

    参数:
        path: 已保存到磁盘的文件路径。
        suffix: 已通过元数据校验的小写扩展名。
    异常:
        HTTPException: 文件内容无效、损坏，或真实格式与扩展名不一致。
    """
    if suffix == ".pdf":
        with path.open("rb") as source:
            if source.read(5) != b"%PDF-":
                raise HTTPException(
                    status_code=415,
                    detail="PDF 文件内容无效",
                )
        return

    expected_format = "PNG" if suffix == ".png" else "JPEG"

    try:
        with Image.open(path) as image:
            if image.format != expected_format:
                raise HTTPException(
                    status_code=415,
                    detail="图片扩展名与真实格式不一致",
                )
            image.verify()
    except (UnidentifiedImageError, OSError):
        raise HTTPException(
            status_code=415,
            detail="图片内容损坏或不是有效图片",
        ) from None


def save_file(file: UploadFile) -> tuple[Path, dict]:
    """校验并分块保存一个上传文件，返回保存路径和文件元数据。

    处理流程：校验元数据、生成唯一存储名、分块写入、限制文件大小、
    检查真实内容。失败时清理本次创建的残留文件，结束时关闭上传文件。

    参数:
        file: FastAPI 接收到的上传文件。
    返回:
        一个元组，包含服务端保存路径和响应所需的文件元数据字典。
    异常:
        HTTPException: 文件为空、大小超限、内容无效或存储文件名冲突。
    """
    suffix = validate_metadata(file)
    stored_name = f"{uuid4().hex}{suffix}"
    target = UPLOAD_DIR / stored_name
    total = 0

    try:
        # xb 表示独占创建；目标已存在时不会覆盖。
        with target.open("xb") as output:
            while chunk := file.file.read(CHUNK_SIZE):
                total += len(chunk)

                if total > MAX_FILE_SIZE:
                    raise HTTPException(
                        status_code=413,
                        detail="单个文件不能超过 10 MiB",
                    )

                output.write(chunk)

        if total == 0:
            raise HTTPException(status_code=400, detail="不能上传空文件")

        validate_content(target, suffix)
    except FileExistsError:
        # 已存在的目标文件不是本例创建的，绝不能删除。
        raise HTTPException(
            status_code=409,
            detail="存储文件名冲突，请重试",
        ) from None
    except Exception:
        # 大小超限、内容无效或磁盘写入失败时清理残留。
        target.unlink(missing_ok=True)
        raise
    finally:
        file.file.close()

    return target, {
        "original_name": Path(file.filename or "").name,
        "stored_name": f"http://localhost:8000/uploads/{stored_name}",
        "content_type": file.content_type,
        "size": total,
    }


@app.post("/upload/single", status_code=201)
def upload_single(file: Annotated[UploadFile, File()]):
    """接收单个上传文件，保存成功后返回文件元数据。

    参数:
        file: 请求中名为 file 的上传文件。
    返回:
        包含成功提示和文件元数据的字典，HTTP 状态码为 201。
    """
    # 保存方法会统一完成校验、写入、内容验证和失败清理。
    _, metadata = save_file(file)
    return {
        "message": "上传成功",
        "file": metadata,
    }

# 挂载文件目录 他必须放在/前面

app.mount(
    "/uploads",
    StaticFiles(directory=str(UPLOAD_DIR)),
    name="uploads",
)

# 挂载 静态目录
app.mount(
    "/",
    StaticFiles(directory=str(BASE_DIR / "static"), html=True),
    name="static",
)



# 启动项目

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000,reload=True)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186

这里使用同步路由 def，FastAPI 会把它放到线程池中执行，磁盘读写和 Pillow 校验不会直接阻塞事件循环。

save_file() 会先检查扩展名和 content_type，再分块写入并限制大小，最后检查真实内容。任何一步失败都会进入清理逻辑。

3. 启动并测试

powershell

uv run uvicorn main:app --reload --host 127.0.0.1 --port 8000

打开交互文档：

text

http://127.0.0.1:8000/docs

找到 POST /upload/single，在 file 字段选择图片或 PDF。

也可以使用 curl.exe。下面以上传 PDF 为例，路径和 MIME 类型必须与真实文件一致：

powershell

curl.exe -X POST http://127.0.0.1:8000/upload/single `
  -F "file=@C:/temp/demo.pdf;type=application/pdf"

上传 PNG 图片时改成：

powershell

curl.exe -X POST http://127.0.0.1:8000/upload/single `
  -F "file=@C:/temp/avatar.png;type=image/png"

成功响应示例：

json

{
  "message": "上传成功",
  "file": {
    "original_name": "avatar.png",
    "stored_name": "6d1f68c5121744f2a43d3cbddfb0e5f2.png",
    "content_type": "image/png",
    "size": 15320
  }
}

再次上传同一个原始文件也会生成新的 stored_name，不会覆盖第一次的文件。

4. 前端核心写法

下面代码假设页面与 FastAPI 接口同源：

html

<form id="single-form">
  <input name="file" type="file" accept=".jpg,.jpeg,.png,.pdf" required />
  <button type="submit">上传</button>
</form>
<pre id="result"></pre>

<script>
  const form = document.querySelector("#single-form");
  const result = document.querySelector("#result");

  form.addEventListener("submit", async (event) => {
    event.preventDefault();

    const body = new FormData(form);
    const response = await fetch("/upload/single", {
      method: "POST",
      body,
    });
    const data = await response.json();

    result.textContent =
      `HTTP ${response.status}\n` + JSON.stringify(data, null, 2);
  });
</script>

不要手动设置 Content-Type: multipart/form-data。浏览器使用 FormData 时会自动携带正确的 boundary。

5. 关于原始文件名也不允许重复

UUID 解决的是“存储名冲突”，不代表业务上的“原始文件名唯一”。如果业务规定同一个用户的原始文件名也不能重复，应当：

在数据库保存 user_id、original_name、stored_name 等元数据。
对 (user_id, original_name) 建唯一约束。
冲突时返回 409 Conflict，并删除刚保存的文件。

只在 Python 列表或内存字典里记录文件名不可靠，因为服务重启或多实例部署后无法保证一致性。

二、多文件上传（图片或者文件）

多文件上传不是把文件路径数组放进 JSON，而是在 multipart 请求中重复使用同一个字段名。本例继续使用第一部分的 main.py、校验函数和保存函数。

1. 本例增加的批量规则

规则	本例约定
文件数量	一次最多 5 个
单文件大小	每个文件仍不能超过 10 MiB
批次总大小	所有文件合计不能超过 20 MiB
文件类型	每个文件分别执行扩展名、MIME 和真实内容检查
批次内原始文件名	不允许重复，重复时返回 409
存储名	每个文件单独生成 UUID，不互相覆盖
失败策略	任意一个失败，本批已经保存的文件全部删除

2. 添加多文件接口

单文件就走 /upload/single 多文件就走/upload/multiple

python

from pathlib import Path
from typing import Annotated
from uuid import uuid4
import uvicorn
from fastapi import FastAPI, File, HTTPException, UploadFile
from PIL import Image, UnidentifiedImageError
from fastapi.staticfiles import StaticFiles
app = FastAPI(title="单文件与多文件上传示例")

BASE_DIR = Path(__file__).resolve().parent
UPLOAD_DIR = BASE_DIR / "uploads"
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)

MAX_FILE_SIZE = 10 * 1024 * 1024
CHUNK_SIZE = 1024 * 1024
MAX_FILES = 5
MAX_TOTAL_SIZE = 20 * 1024 * 1024

ALLOWED_TYPES = {
    ".jpg": {"image/jpeg"},
    ".jpeg": {"image/jpeg"},
    ".png": {"image/png"},
    ".pdf": {"application/pdf"},
}


def validate_metadata(file: UploadFile) -> str:
    """校验上传文件的名称、扩展名和客户端声明的 MIME 类型。

    参数:
        file: FastAPI 接收到的上传文件。
    返回:
        通过校验的小写文件扩展名，例如 ".jpg"。
    异常:
        HTTPException: 扩展名不受支持，或 MIME 类型与扩展名不匹配。
    说明:
        MIME 类型由客户端提供，因此这里只做初步检查，还需验证真实文件内容。
    """
    original_name = Path(file.filename or "").name
    suffix = Path(original_name).suffix.lower()

    if suffix not in ALLOWED_TYPES:
        raise HTTPException(
            status_code=415,
            detail="只支持 JPG、JPEG、PNG 和 PDF 文件",
        )

    if file.content_type not in ALLOWED_TYPES[suffix]:
        raise HTTPException(
            status_code=415,
            detail="文件扩展名与 content_type 不匹配",
        )

    return suffix


def validate_content(path: Path, suffix: str) -> None:
    """检查已保存文件的真实内容，防止伪造扩展名或 MIME 类型。

    PDF 通过文件头进行初步验证；JPG、JPEG 和 PNG 使用 Pillow 核对真实
    格式并检查文件完整性。

    参数:
        path: 已保存到磁盘的文件路径。
        suffix: 已通过元数据校验的小写扩展名。
    异常:
        HTTPException: 文件内容无效、损坏，或真实格式与扩展名不一致。
    """
    if suffix == ".pdf":
        with path.open("rb") as source:
            if source.read(5) != b"%PDF-":
                raise HTTPException(
                    status_code=415,
                    detail="PDF 文件内容无效",
                )
        return

    expected_format = "PNG" if suffix == ".png" else "JPEG"

    try:
        with Image.open(path) as image:
            if image.format != expected_format:
                raise HTTPException(
                    status_code=415,
                    detail="图片扩展名与真实格式不一致",
                )
            image.verify()
    except (UnidentifiedImageError, OSError):
        raise HTTPException(
            status_code=415,
            detail="图片内容损坏或不是有效图片",
        ) from None


def save_file(file: UploadFile) -> tuple[Path, dict]:
    """校验并分块保存一个上传文件，返回保存路径和文件元数据。

    处理流程：校验元数据、生成唯一存储名、分块写入、限制文件大小、
    检查真实内容。失败时清理本次创建的残留文件，结束时关闭上传文件。

    参数:
        file: FastAPI 接收到的上传文件。
    返回:
        一个元组，包含服务端保存路径和响应所需的文件元数据字典。
    异常:
        HTTPException: 文件为空、大小超限、内容无效或存储文件名冲突。
    """
    suffix = validate_metadata(file)
    stored_name = f"{uuid4().hex}{suffix}"
    target = UPLOAD_DIR / stored_name
    total = 0

    try:
        # xb 表示独占创建；目标已存在时不会覆盖。
        with target.open("xb") as output:
            while chunk := file.file.read(CHUNK_SIZE):
                total += len(chunk)

                if total > MAX_FILE_SIZE:
                    raise HTTPException(
                        status_code=413,
                        detail="单个文件不能超过 10 MiB",
                    )

                output.write(chunk)

        if total == 0:
            raise HTTPException(status_code=400, detail="不能上传空文件")

        validate_content(target, suffix)
    except FileExistsError:
        # 已存在的目标文件不是本例创建的，绝不能删除。
        raise HTTPException(
            status_code=409,
            detail="存储文件名冲突，请重试",
        ) from None
    except Exception:
        # 大小超限、内容无效或磁盘写入失败时清理残留。
        target.unlink(missing_ok=True)
        raise
    finally:
        file.file.close()

    return target, {
        "original_name": Path(file.filename or "").name,
        "stored_name": f"http://localhost:8000/uploads/{stored_name}",
        "content_type": file.content_type,
        "size": total,
    }


@app.post("/upload/single", status_code=201)
def upload_single(file: Annotated[UploadFile, File()]):
    """接收单个上传文件，保存成功后返回文件元数据。

    参数:
        file: 请求中名为 file 的上传文件。
    返回:
        包含成功提示和文件元数据的字典，HTTP 状态码为 201。
    """
    # 保存方法会统一完成校验、写入、内容验证和失败清理。
    _, metadata = save_file(file)
    return {
        "message": "上传成功",
        "file": metadata,
    }


# 多文件上传

@app.post("/upload/multiple", status_code=201)
def upload_multiple(files: Annotated[list[UploadFile], File()]):
    if not files:
        raise HTTPException(status_code=400, detail="请至少选择一个文件")

    if len(files) > MAX_FILES:
        raise HTTPException(
            status_code=400,
            detail="一次最多上传 5 个文件",
        )

    original_names = [Path(file.filename or "").name for file in files]
    if len(original_names) != len(set(original_names)):
        raise HTTPException(
            status_code=409,
            detail="同一批次中不能有重复的原始文件名",
        )

    saved: list[tuple[Path, dict]] = []
    total_size = 0

    try:
        for file in files:
            target, metadata = save_file(file)
            saved.append((target, metadata))

            total_size += metadata["size"]
            if total_size > MAX_TOTAL_SIZE:
                raise HTTPException(
                    status_code=413,
                    detail="本批文件总大小不能超过 20 MiB",
                )
    except Exception:
        # 当前文件由 save_file 自己清理，这里回滚之前已完成的文件。
        for target, _ in saved:
            target.unlink(missing_ok=True)
        raise

    return {
        "message": "批量上传成功",
        "count": len(saved),
        "total_size": total_size,
        "files": [metadata for _, metadata in saved],
    }

# 挂载文件目录 他必须放在/前面

app.mount(
    "/uploads",
    StaticFiles(directory=str(UPLOAD_DIR)),
    name="uploads",
)

# 挂载 静态目录
app.mount(
    "/",
    StaticFiles(directory=str(BASE_DIR / "static"), html=True),
    name="static",
)



# 启动项目

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000,reload=True)

这里有两层清理：

save_file() 清理当前正在写入但校验失败的文件。
upload_multiple() 删除本批之前已经保存成功的文件。

因此，假设第 3 个文件类型不正确，前两个文件也会被删除，最终不会出现“接口报错但磁盘只保存了一半”的情况。

3. 前端多文件写法

input 需要添加 multiple，并将每个文件都追加到名为 files 的字段中：

html

<form id="multiple-form">
  <input
    name="files"
    type="file"
    accept=".jpg,.jpeg,.png,.pdf"
    multiple
    required
  />
  <button type="submit">批量上传</button>
</form>
<pre id="multiple-result"></pre>

<script>
  const form = document.querySelector("#multiple-form");
  const result = document.querySelector("#multiple-result");

  form.addEventListener("submit", async (event) => {
    event.preventDefault();

    const input = form.querySelector('input[type="file"]');
    const body = new FormData();

    for (const file of input.files) {
      body.append("files", file);
    }

    const response = await fetch("/upload/multiple", {
      method: "POST",
      body,
    });
    const data = await response.json();

    result.textContent =
      `HTTP ${response.status}\n` + JSON.stringify(data, null, 2);
  });
</script>

前端还可以提前检查 input.files.length 是否大于 5，但这只能改善交互，后端限制仍然必须保留。

4. 命令行测试

下面在同一请求中上传一张 PNG 和一个 PDF，注意两个部分都使用 files：

powershell

curl.exe -X POST http://127.0.0.1:8000/upload/multiple `
  -F "files=@C:/temp/avatar.png;type=image/png" `
  -F "files=@C:/temp/demo.pdf;type=application/pdf"

建议继续测试：

一次上传 6 个文件，应返回数量限制错误。
同一批次重复添加两个同名文件，应返回 409。
第二个文件使用伪造扩展名，应返回 415，且第一个文件也会被回滚。
让总大小超过 20 MiB，应返回 413，且不保留本批文件。

5. 应用内校验不是唯一的大小防线

FastAPI 开始执行路由函数前，multipart 解析可能已经接收并缓存了请求内容。因此，除了本例的应用内检查，生产环境还应在反向代理或 ASGI 接收层限制请求体大小，并限制并发上传数和临时磁盘空间。

三、文件上传到阿里云 OSS

阿里云 OSS 是对象存储。这里的上传链路是：

text

浏览器 → FastAPI → 阿里云 OSS

AccessKey 只保存在后端，绝不能写进 HTML、JavaScript 或前端环境变量中。本例仍然在 FastAPI 中检查大小、类型和真实内容，验证通过后再调用 OSS。

1. 准备 Bucket、权限和配置

先在阿里云完成以下准备：

创建一个 OSS Bucket，读取权限建议保持私有。
确认 Bucket 所在地域及对应 Endpoint，例如杭州地域的 Endpoint 格式为 https://oss-cn-hangzhou.aliyuncs.com。
使用权限受限的 RAM 身份，只授予指定 Bucket 或指定前缀所需的上传权限，不要使用拥有全部云资源权限的账号。
准备 AccessKey ID 和 AccessKey Secret。

本例使用独立目录：

text

oss-demo/
├─ main.py
└─ .env

安装依赖：

powershell

uv add python-multipart pillow oss2 python-dotenv

在 .env 中填写配置。下面所有需要替换的值都先用中文占位：

dotenv

OSS_ACCESS_KEY_ID=请填写你的阿里云AccessKeyID
OSS_ACCESS_KEY_SECRET=请填写你的阿里云AccessKey密钥
OSS_ENDPOINT=请填写你的OSS地域Endpoint
OSS_BUCKET=请填写你的Bucket名称

配置项	应填写的内容
`OSS_ACCESS_KEY_ID`	受限 RAM 身份的 AccessKey ID
`OSS_ACCESS_KEY_SECRET`	与 AccessKey ID 对应的密钥
`OSS_ENDPOINT`	Bucket 所在地域的 OSS Endpoint
`OSS_BUCKET`	实际创建的 Bucket 名称

不要把真实 .env 提交到 Git。下面的代码会检测空值和“请填写”开头的内容，未替换占位时会直接提示缺少配置。

2. 后端完整代码

将下面代码保存为 OSS 示例目录中的 main.py：

python

import os
from io import BytesIO
from pathlib import Path
from typing import Annotated
from uuid import uuid4

import oss2
from dotenv import load_dotenv
from fastapi import FastAPI, File, HTTPException, UploadFile
from oss2.exceptions import OssError
from PIL import Image, UnidentifiedImageError

load_dotenv()


def required_env(name: str) -> str:
    value = os.getenv(name)
    if not value or value.startswith("请填写"):
        raise RuntimeError(f"请先在 .env 中正确配置 {name}")
    return value


ACCESS_KEY_ID = required_env("OSS_ACCESS_KEY_ID")
ACCESS_KEY_SECRET = required_env("OSS_ACCESS_KEY_SECRET")
ENDPOINT = required_env("OSS_ENDPOINT")
BUCKET_NAME = required_env("OSS_BUCKET")

app = FastAPI(title="FastAPI 上传到阿里云 OSS")

bucket = oss2.Bucket(
    oss2.Auth(ACCESS_KEY_ID, ACCESS_KEY_SECRET),
    ENDPOINT,
    BUCKET_NAME,
)

MAX_FILE_SIZE = 10 * 1024 * 1024
ALLOWED_TYPES = {
    ".jpg": {"image/jpeg"},
    ".jpeg": {"image/jpeg"},
    ".png": {"image/png"},
    ".pdf": {"application/pdf"},
}


def validate_metadata(file: UploadFile) -> str:
    original_name = Path(file.filename or "").name
    suffix = Path(original_name).suffix.lower()

    if suffix not in ALLOWED_TYPES:
        raise HTTPException(
            status_code=415,
            detail="只支持 JPG、JPEG、PNG 和 PDF 文件",
        )

    if file.content_type not in ALLOWED_TYPES[suffix]:
        raise HTTPException(
            status_code=415,
            detail="文件扩展名与 content_type 不匹配",
        )

    return suffix


def validate_content(content: bytes, suffix: str) -> None:
    if suffix == ".pdf":
        if not content.startswith(b"%PDF-"):
            raise HTTPException(
                status_code=415,
                detail="PDF 文件内容无效",
            )
        return

    expected_format = "PNG" if suffix == ".png" else "JPEG"

    try:
        with Image.open(BytesIO(content)) as image:
            if image.format != expected_format:
                raise HTTPException(
                    status_code=415,
                    detail="图片扩展名与真实格式不一致",
                )
            image.verify()
    except (UnidentifiedImageError, OSError):
        raise HTTPException(
            status_code=415,
            detail="图片内容损坏或不是有效图片",
        ) from None


@app.post("/upload/oss", status_code=201)
def upload_to_oss(file: Annotated[UploadFile, File()]):
    suffix = validate_metadata(file)

    try:
        # 本例限制为 10 MiB，因此最多读取限制大小加 1 字节。
        content = file.file.read(MAX_FILE_SIZE + 1)

        if not content:
            raise HTTPException(status_code=400, detail="不能上传空文件")

        if len(content) > MAX_FILE_SIZE:
            raise HTTPException(
                status_code=413,
                detail="单个文件不能超过 10 MiB",
            )

        # 先验证，再上传，避免无效文件进入 OSS。
        validate_content(content, suffix)
        object_key = f"uploads/{uuid4().hex}{suffix}"

        try:
            result = bucket.put_object(
                object_key,
                content,
                headers={
                    "Content-Type": file.content_type,
                    # 即使发生对象名冲突，也不覆盖已有对象。
                    "x-oss-forbid-overwrite": "true",
                },
            )
        except OssError as exc:
            if exc.status == 409:
                raise HTTPException(
                    status_code=409,
                    detail="OSS 对象名冲突，请重试",
                ) from exc

            raise HTTPException(
                status_code=502,
                detail=f"阿里云 OSS 上传失败：{exc.message}",
            ) from exc

        return {
            "message": "上传到阿里云 OSS 成功",
            "original_name": Path(file.filename or "").name,
            "object_key": object_key,
            "size": len(content),
            "etag": result.etag,
        }
    finally:
        file.file.close()

关键点如下：

object_key 是 OSS 中的对象名，不是本地磁盘路径。本例使用 uploads/UUID.扩展名。
x-oss-forbid-overwrite: true 用来禁止覆盖已有对象；UUID 则让对象名冲突的概率极低。
Content-Type 会作为对象元数据保存，后续下载或预览时可以返回正确的类型。
put_object() 是网络操作。这里使用同步路由，让 FastAPI 在线程池中执行，避免直接阻塞事件循环。
返回 object_key 和 etag 即可，不要因为需要展示文件就擅自把 Bucket 改成公共读。

3. 启动并测试

先确认 .env 中的四个中文占位都已替换，再启动：

powershell

uv run uvicorn main:app --reload --host 127.0.0.1 --port 8000

可以在 /docs 的 POST /upload/oss 中选择文件，也可以使用命令行。把中文路径占位换成真实图片：

powershell

curl.exe -X POST http://127.0.0.1:8000/upload/oss `
  -F "file=@请填写本地PNG图片路径;type=image/png"

成功响应示例：

json

{
  "message": "上传到阿里云 OSS 成功",
  "original_name": "avatar.png",
  "object_key": "uploads/55c0a004a87949c4b1882d73a8d2a584.png",
  "size": 15320,
  "etag": "\"示例ETag值\""
}

前端可以继续使用第一部分的上传表单，只需把请求地址改为：

javascript

const response = await fetch("/upload/oss", {
  method: "POST",
  body: new FormData(form),
});

前端只负责把文件交给自己的 FastAPI 后端，不接触 AccessKey，也不直接管理 OSS 权限。

4. 常见 OSS 错误

错误现象	优先检查
启动时提示需要配置环境变量	`.env` 是否与 `main.py` 位于同一目录，中文占位是否全部替换
`InvalidAccessKeyId`	AccessKey ID 是否正确、是否已禁用或删除
`SignatureDoesNotMatch`	AccessKey Secret 是否正确，配置中是否混入多余空格
`NoSuchBucket`	Bucket 名称是否正确，Endpoint 是否属于 Bucket 所在地域
`AccessDenied`	RAM 身份是否有该 Bucket 和 `uploads/` 前缀的上传权限
返回 413 或 415	文件是否超过大小限制，扩展名、MIME 与真实内容是否一致
返回 409	存储名或对象名冲突，应重新生成名称，不要改成覆盖上传

5. 大文件与生产环境注意事项

本例最多读取 10 MiB，因此用 bytes 调用 put_object() 便于教学。大文件不要照搬为无限读取内存，应改用流式上传、OSS 分片上传，或由后端签发临时凭证让客户端直传。
即使 OSS 上传成功，也应在数据库保存 object_key、原始文件名、大小、上传者和业务归属，后续通过权限检查决定谁能下载。
私有 Bucket 的下载可使用后端代理或有时效的签名 URL，不要提供任意对象名读取接口。
AccessKey 应定期轮换。生产部署优先考虑实例角色、临时凭证等方案，避免长期密钥散落在代码和部署脚本中。
反向代理的请求体限制、上传并发限制、病毒扫描和内容安全审查仍然需要按业务配置。

12. FastAPI 文件上传 ​

一、单文件上传（图片或者文件） ​

1. 本例的上传规则 ​

2. 后端完整代码 ​

3. 启动并测试 ​

4. 前端核心写法 ​

5. 关于原始文件名也不允许重复 ​

二、多文件上传（图片或者文件） ​

1. 本例增加的批量规则 ​

2. 添加多文件接口 ​

3. 前端多文件写法 ​

4. 命令行测试 ​

5. 应用内校验不是唯一的大小防线 ​

三、文件上传到阿里云 OSS ​

1. 准备 Bucket、权限和配置 ​

2. 后端完整代码 ​

3. 启动并测试 ​

4. 常见 OSS 错误 ​

5. 大文件与生产环境注意事项 ​

12. FastAPI 文件上传

一、单文件上传（图片或者文件）

1. 本例的上传规则

2. 后端完整代码

3. 启动并测试

4. 前端核心写法

5. 关于原始文件名也不允许重复

二、多文件上传（图片或者文件）

1. 本例增加的批量规则

2. 添加多文件接口

3. 前端多文件写法

4. 命令行测试

5. 应用内校验不是唯一的大小防线

三、文件上传到阿里云 OSS

1. 准备 Bucket、权限和配置

2. 后端完整代码

3. 启动并测试

4. 常见 OSS 错误

5. 大文件与生产环境注意事项