FastAPI: 文件上传与下载 — 大文件与批量导入
最后更新：2026-08-26
文件上传像快递收件——小包裹直接签收（内存），大件物流需要分批卸货（流式写入）；下载像发货——单件快递（FileResponse），批量发货走传送带（StreamingResponse）。
1. 你将学到
UploadFile与File()参数：流式上传 vs 内存上传- 大文件处理：分块读取、
shutil.copyfileobj流式写入 - CSV/Excel 解析：
pandas+ Pydantic 模型联合验证 - 文件响应：
FileResponse/StreamingResponse下载端点 - Alice 场景：Bob 上传包含百万行价格的 CSV 文件 → 后台 Celery 异步解析 → 导入完成后通知
2. Alice 的真实故事
(1) 痛点：百万行 CSV 导入内存溢出
Bob 每月上传一个包含百万行价格的 CSV 文件到 PriceTracker，之前的实现把整个文件读入内存再解析，1GB 的 CSV 文件导致 Python 进程 OOM（Out of Memory）崩溃。更糟的是，CSV 中混杂了非法数据（负数价格、无效币种），导入一半失败后数据库处于不一致状态。
(2) 流式上传+异步处理的解法
FastAPI 的 UploadFile 默认将文件存为临时文件（不占内存），配合 Celery 异步处理百万行解析，Pydantic 逐行校验跳过非法数据，事务化批量插入确保一致性。
(3) 收益
1GB CSV 导入从 OOM 崩溃变成稳定运行，内存占用从 2GB 降到 50MB。Celery Worker 处理百万行约 10 分钟，Bob 可通过任务状态端点实时查看进度。
3. UploadFile 基础
(1) UploadFile vs bytes
▶ 示例：简单文件上传
PYTHON
from fastapi import FastAPI, UploadFile, File, HTTPException
app = FastAPI()
@app.post("/upload/single")
async def upload_single(file: UploadFile = File(...)):
# UploadFile: file stored as temp file, NOT in memory
content = await file.read()
return {
"filename": file.filename,
"size": len(content),
"content_type": file.content_type,
}
输出：
TEXT
📖 仅展示
# 函数定义成功
| 方式 | 内存占用 | 适用文件大小 | API |
|---|---|---|---|
bytes |
全部加载到内存 | < 2MB | file: bytes = File() |
UploadFile |
临时文件（流式） | 无限制 | file: UploadFile = File() |
▶ 示例：多文件上传
PYTHON
@app.post("/upload/multiple")
async def upload_multiple(files: list[UploadFile] = File(...)):
results = []
for file in files:
content = await file.read()
results.append({
"filename": file.filename,
"size": len(content),
})
return {"uploaded": len(results), "files": results}
输出：
TEXT
📖 仅展示
# 函数定义成功
4. 大文件流式处理
(1) 分块读取与流式写入
▶ 示例：大文件流式保存
PYTHON
import shutil
from pathlib import Path
from fastapi import FastAPI, UploadFile, File
app = FastAPI()
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
@app.post("/upload/large")
async def upload_large_file(file: UploadFile = File(...)):
# Stream file to disk - never loads entire file into memory
dest = UPLOAD_DIR / file.filename
with open(dest, "wb") as buffer:
# Copy in chunks (default 64KB buffer)
shutil.copyfileobj(file.file, buffer)
file_size = dest.stat().st_size
return {
"filename": file.filename,
"size_bytes": file_size,
"saved_to": str(dest),
}
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：自定义分块大小
PYTHON
CHUNK_SIZE = 1024 * 1024 # 1MB chunks
@app.post("/upload/chunked")
async def upload_chunked(file: UploadFile = File(...)):
dest = UPLOAD_DIR / file.filename
bytes_written = 0
with open(dest, "wb") as buffer:
while chunk := await file.read(CHUNK_SIZE):
buffer.write(chunk)
bytes_written += len(chunk)
return {"filename": file.filename, "bytes_written": bytes_written}
输出：
TEXT
📖 仅展示
# 函数定义成功
5. CSV/Excel 批量导入
(1) 大文件上传+异步处理流程
sequenceDiagram
participant Bob as Bob Frontend
participant API as FastAPI
participant Temp as Temp File
participant Celery as Celery Worker
participant DB as PostgreSQL
Bob->>API: POST /import/csv (UploadFile)
API->>Temp: Save to temp file
API-->>Bob: 202 Accepted + task_id
API->>Celery: Trigger parse task
Celery->>Temp: Read CSV in chunks
Celery->>Celery: Validate each row (Pydantic)
Celery->>DB: Batch insert valid rows
Celery-->>Celery: Report progress
Bob->>API: GET /tasks/{task_id}
API-->>Bob: Progress: 75%
Celery-->>Celery: Task complete
Bob->>API: GET /tasks/{task_id}
API-->>Bob: Status: SUCCESS, imported: 950000
▶ 示例：CSV 上传端点 + Celery 任务触发
PYTHON
import csv
import io
from fastapi import FastAPI, UploadFile, File, Depends
from app.tasks import import_prices_from_csv
app = FastAPI()
UPLOAD_DIR = Path("uploads")
@app.post("/api/v1/import/csv")
async def import_csv(
file: UploadFile = File(..., description="CSV file with price data"),
user=Depends(require_subscription("pro")),
):
if not file.filename.endswith(".csv"):
raise HTTPException(status_code=400, detail="Only CSV files accepted")
# Save uploaded file
dest = UPLOAD_DIR / f"{uuid4()}.csv"
with open(dest, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# Trigger Celery task for async processing
task = import_prices_from_csv.delay(str(dest), user_id=user.id)
return {
"task_id": task.id,
"filename": file.filename,
"status": "processing",
"message": "File uploaded. Check task status for progress.",
}
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：Celery 任务逐行解析 CSV
PYTHON
# app/tasks/import_tasks.py
from app.core.celery_app import celery_app
from pydantic import BaseModel, Field, field_validator
import csv
class PriceRow(BaseModel):
product_id: int = Field(gt=0)
price: float = Field(gt=0)
currency: str = Field(default="USD", pattern=r"^[A-Z]{3}$")
source: str = Field(max_length=100, default="csv_import")
@field_validator("price")
@classmethod
def round_price(cls, v: float) -> float:
return round(v, 2)
@celery_app.task(bind=True)
def import_prices_from_csv(self, file_path: str, user_id: int):
valid_rows = []
invalid_rows = []
total_rows = 0
with open(file_path, "r") as f:
reader = csv.DictReader(f)
for row in reader:
total_rows += 1
try:
validated = PriceRow(**row)
valid_rows.append(validated.model_dump())
except Exception as e:
invalid_rows.append({"row": total_rows, "error": str(e)})
# Report progress every 10000 rows
if total_rows % 10000 == 0:
self.update_state(
state="PROGRESS",
meta={"current": total_rows, "valid": len(valid_rows), "invalid": len(invalid_rows)},
)
# Batch insert valid rows
batch_insert_prices(valid_rows, batch_size=5000)
# Clean up temp file
Path(file_path).unlink(missing_ok=True)
return {
"total": total_rows,
"imported": len(valid_rows),
"skipped": len(invalid_rows),
}
输出：
TEXT
📖 仅展示
# 函数定义成功
(2) 文件格式处理矩阵
| 格式 | 解析库 | 优点 | 缺点 |
|---|---|---|---|
| CSV | csv / pandas | 轻量、流式 | 无类型、编码问题 |
| XLSX | openpyxl / pandas | 有类型、多 Sheet | 内存占用高 |
| JSON | json / orjson | 结构化、Pydantic 友好 | 文件较大 |
| Parquet | pyarrow | 列式存储、压缩率高 | 需额外库 |
6. 文件下载
(1) FileResponse 与 StreamingResponse
▶ 示例：FileResponse 下载文件
PYTHON
from fastapi import FastAPI
from fastapi.responses import FileResponse
from pathlib import Path
app = FastAPI()
@app.get("/download/prices/csv")
async def download_prices_csv(
user=Depends(require_subscription("pro")),
):
# Generate CSV file (or use pre-generated)
file_path = Path("exports/prices.csv")
return FileResponse(
path=file_path,
filename="price_data.csv",
media_type="text/csv",
)
输出：
TEXT
📖 仅展示
# 函数定义成功
▶ 示例：StreamingResponse 流式生成 CSV
PYTHON
from fastapi.responses import StreamingResponse
import csv
import io
from app.core.deps import get_db
@app.get("/api/v1/export/prices")
async def export_prices(
category: str | None = None,
db: AsyncSession = Depends(get_db),
user=Depends(require_subscription("pro")),
):
async def generate_csv():
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(["product_id", "name", "price", "currency", "recorded_at"])
yield output.getvalue()
output.seek(0)
output.truncate(0)
# Stream rows in batches
offset = 0
batch_size = 5000
while True:
rows = await fetch_price_batch(db, category, offset, batch_size)
if not rows:
break
for row in rows:
writer.writerow([
row.product_id, row.name,
row.price, row.currency, row.recorded_at,
])
yield output.getvalue()
output.seek(0)
output.truncate(0)
offset += batch_size
return StreamingResponse(
generate_csv(),
media_type="text/csv",
headers={"Content-Disposition": "attachment; filename=prices.csv"},
)
输出：
TEXT
📖 仅展示
# 函数定义成功
| 响应类型 | 适用场景 | 内存占用 |
|---|---|---|
FileResponse |
已有文件 | 低（OS 层面流式） |
StreamingResponse |
动态生成 | 极低（逐行生成） |
❓ 常见问题
Q UploadFile 的临时文件什么时候删除？
A 请求结束后自动删除。如需保留，在上传处理中将其复制到持久目录。
Q 上传文件大小有限制吗？
A FastAPI 无框架限制，但 Uvicorn 默认限制请求体大小。生产环境用 Nginx 的
client_max_body_size 控制。Q CSV 导入中部分行无效怎么办？
A 策略取决于业务——严格模式（任何无效行全部回滚）或宽松模式（跳过无效行，导入有效行）。PriceTracker 用宽松模式，返回导入数和跳过数。
Q Excel 文件怎么处理？
A 用 pandas 的
read_excel() 解析，但 XLSX 不能流式读取（必须全加载）。百万行建议转 CSV 格式上传。Q StreamingResponse 的 generator 必须是 async 吗？
A 不必须，同步 generator 也可以。但 async generator 不会阻塞事件循环，推荐使用。
Q 如何限制上传文件类型？
A 检查
file.content_type（如 text/csv）和文件扩展名。注意 content_type 可被伪造，仅做前端提示，服务端仍需解析校验。📖 小节
UploadFile流式存储到临时文件，不占内存，适合任意大小文件- 大文件用
shutil.copyfileobj()或分块read()流式写入磁盘 - CSV 上传 + Celery 异步解析：API 立即返回 task_id，后台逐行校验+批量插入
- Pydantic 逐行校验 CSV 数据，无效行跳过并记录，宽松模式保证部分导入
FileResponse下载已有文件，StreamingResponse动态生成并流式返回
📝 作业
- 基础题（难度⭐）：实现 CSV 上传端点，接收 UploadFile，读取内容并返回行数和列名。提示：
file: UploadFile = File(...)+csv.DictReader(io.StringIO(content)) - 进阶题（难度⭐⭐）：实现 CSV 上传 → Celery 异步解析流程：上传文件保存到临时目录，触发 Celery 任务，返回 task_id，另一个端点查询任务状态和导入结果。提示：
shutil.copyfileobj+import_prices.delay(file_path) - 挑战题（难度⭐⭐⭐）：实现 StreamingResponse 导出端点：从数据库流式读取价格数据，逐行生成 CSV 返回，内存占用不超过 1MB，支持按品类过滤。提示：
async def generate()+yield+StreamingResponse(generate(), media_type="text/csv")
---|
