钟掌握 FastAPI 文件上传：从入门到进阶的完整指南

一、FastAPI 文件上传的核心优势

FastAPI 作为现代 Python Web 框架的代表，其文件上传功能具有显著优势。基于 Starlette 的异步架构，FastAPI 能够高效处理并发文件上传请求，尤其在处理大文件或多文件并行上传时，性能表现远超传统同步框架。其类型提示机制与自动生成的 OpenAPI 文档，使开发者能够快速构建符合 RESTful 规范的上传接口。

核心优势体现在三方面：其一，异步处理能力通过 async/await 模式释放 I/O 阻塞，提升吞吐量；其二，自动化的请求体解析支持单文件、多文件及混合表单数据的无缝处理；其三，与 Pydantic 模型的无缝集成，可对上传文件进行严格的类型与格式校验。例如，通过 UploadFile 对象可直接访问文件名、内容类型及文件指针，避免手动解析请求体的繁琐操作。

二、基础文件上传实现

1. 单文件上传

基础实现需依赖 FastAPI 的 File 依赖项与 UploadFile 模型。以下代码展示了一个完整的单文件上传接口：

from fastapi import FastAPI, UploadFile, File
app = FastAPI()
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    # 此处可添加文件存储逻辑
    return {"filename": file.filename, "size": len(contents)}

关键点解析：File(...) 表示该参数为必传文件字段；UploadFile 对象提供 read() 方法异步读取文件内容；返回结果包含文件名与字节长度，便于前端验证。

2. 多文件上传

处理多文件需结合 List 类型与 UploadFile：

from typing import List
@app.post("/multi-upload/")
async def upload_files(files: List[UploadFile] = File(...)):
    results = []
    for file in files:
        contents = await file.read()
        results.append({
            "filename": file.filename,
            "size": len(contents)
        })
    return results

此实现支持前端通过 multipart/form-data 格式同时上传多个文件，服务端通过循环处理每个文件对象。

三、安全控制与校验

1. 文件类型限制

通过检查 UploadFile.content_type 属性可限制允许的文件类型：

ALLOWED_TYPES = {"image/jpeg", "image/png"}
@app.post("/secure-upload/")
async def secure_upload(file: UploadFile = File(...)):
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(status_code=400, detail="Invalid file type")
    # 继续处理

2. 文件大小限制

FastAPI 默认不限制文件大小，需通过中间件或手动校验实现。以下示例限制文件为 10MB 以内：

MAX_SIZE = 10 * 1024 * 1024  # 10MB
@app.post("/size-limited/")
async def size_limited_upload(file: UploadFile = File(...)):
    contents = await file.read()
    if len(contents) > MAX_SIZE:
        raise HTTPException(status_code=400, detail="File too large")
    return {"size": len(contents)}

3. 病毒扫描集成

生产环境建议集成 ClamAV 等开源病毒扫描工具。可通过子进程调用扫描命令，或使用 pyclamd 库实现：

import pyclamd
cd = pyclamd.ClamdUnixSocket()
@app.post("/scan-upload/")
async def scan_upload(file: UploadFile = File(...)):
    contents = await file.read()
    # 临时保存文件供扫描
    with open("temp_file", "wb") as f:
        f.write(contents)
    scan_result = cd.scan_file("temp_file")
    if scan_result and "FOUND" in scan_result:
        raise HTTPException(status_code=400, detail="Virus detected")
    return {"status": "clean"}

四、性能优化策略

1. 流式上传处理

对于大文件，采用流式读取避免内存溢出：

@app.post("/stream-upload/")
async def stream_upload(file: UploadFile = File(...)):
    chunk_size = 1024 * 1024  # 1MB 块
    total_size = 0
    while True:
        chunk = await file.read(chunk_size)
        if not chunk:
            break
        total_size += len(chunk)
        # 处理每个数据块（如写入磁盘或云存储）
    return {"total_size": total_size}

2. 异步存储集成

结合 aiofiles 实现异步文件写入：

import aiofiles
@app.post("/async-save/")
async def async_save(file: UploadFile = File(...)):
    async with aiofiles.open(f"uploads/{file.filename}", "wb") as f:
        while chunk := await file.read(1024 * 1024):  # 1MB 块
            await f.write(chunk)
    return {"status": "saved"}

3. 云存储集成

以 AWS S3 为例，通过 boto3 异步客户端上传：

import boto3
from botocore.config import Config
s3_config = Config(
    max_pool_connections=10,
    retries={"max_attempts": 3}
)
s3_client = boto3.client("s3", config=s3_config)
@app.post("/s3-upload/")
async def s3_upload(file: UploadFile = File(...)):
    contents = await file.read()
    s3_client.put_object(
        Bucket="my-bucket",
        Key=file.filename,
        Body=contents
    )
    return {"s3_key": file.filename}

五、高级场景应用

1. 分片上传实现

支持大文件分片上传需前端配合，服务端需校验分片顺序与完整性：

@app.post("/chunk-upload/")
async def chunk_upload(
    file: UploadFile = File(...),
    chunk_index: int = Form(...),
    total_chunks: int = Form(...)
):
    chunk_data = await file.read()
    # 将分片数据按顺序写入临时文件
    with open(f"temp_{total_chunks}.part", "ab") as f:
        f.write(chunk_data)
    # 所有分片上传完成后合并
    if chunk_index == total_chunks - 1:
        # 合并逻辑...
        return {"status": "complete"}
    return {"status": "chunk_saved"}

2. 元数据关联

通过 Pydantic 模型关联文件元数据：

from pydantic import BaseModel
class FileMeta(BaseModel):
    user_id: str
    description: str
@app.post("/meta-upload/")
async def meta_upload(
    file: UploadFile = File(...),
    meta: FileMeta = Body(...)
):
    # 存储文件与元数据
    return {"meta": meta, "filename": file.filename}

六、最佳实践总结

1. 安全优先：始终校验文件类型、大小，并集成病毒扫描。
2. 异步到底：利用 async/await 释放 I/O 阻塞，提升并发能力。
3. 流式处理：大文件采用分块读取，避免内存爆炸。
4. 云存储集成：优先使用对象存储服务（如 S3、MinIO）替代本地存储。
5. 监控告警：记录上传失败率、平均耗时等指标，设置阈值告警。

通过以上策略，开发者可构建出高性能、高可用的 FastAPI 文件上传服务，满足从个人项目到企业级应用的多样化需求。