做后端这行。迟早会撞上一道绕不开的坎——使用者要传一个几 GB 甚至几十 GB 的文件，网络还时断时续。如果还用最朴素的 UploadFile 一把梭。内存爆掉是轻的，使用者传到一半网络一抖直接从头再来才是真的崩溃。下面把 FastAPI 场景下处理大文件的几条主流路线捋一遍，并给出一套能直接抄作业的断点续传实现。

大文件传输的几种技术方向

面对大文件。业界基本化出了三条路子，各有各的适用场景。

第一条路是让 FastAPI 亲自上阵。用流式读写扛住大文件，服务端一边接收一边写盘，全程不把整个文件塞进内存。这条路简单直接，适合中等规模、自建存储的场景。用异步文件 IO配合合理的分块大小就能跑得很稳。要命的是如果文件真的很大，普通的 python-multipart 解析方式和客户端库都得换成支持流式传输的版本。否则内存和超时问题照样找上门。说起来，

第二条路是把 FastAPI 从传输链路里请出去。只负责发一张预签名 URL真正的大文件数据直接从客户端怼到对象存储上。基准测试表明，对于真正的大文件。最快的方式就是干脆绕过 FastAPI，让客户端直连对象存储。服务端压力几乎为零，代码里可以看到完整实现范例：FastAPI 生成签名、MongoDB 记录上传状态、S3 承载实际数据。说起来，

第三条路是搞一套标准化的断点续传协议。代表就是. 它是基于 HTTP 的开放协议，专为可恢复上传设计，客户端和服务端都有现成实现，社区已经有人把它移植到 FastAPI。如果你的产品需要跨网站、跨客户端统一的续传体验，走 tus 能省掉大量重复造轮子的工作。

存储后端怎么选

S3 的 Multipart Upload 机制本身就是为大文件而生——先发起上传任务拿到upload_id。 再把文件切成多个分片并发上传，每片返回Etag,最终一次调用合并接口把所有分片拼成完整对象。这套机制天然带有断点续传特性：哪个分片失败。只重传那一片即可，无需从头开始。

断点续传的主要原理

The problem can be split into two distinct parts:

上传方向这方面，分片 + 状态记录

The core idea is to cut large file into fixed‑size chunks,upload each chunk independently,and let server record which chunks have already been received.

If total file size is $S$。and each chunk size is $C$,n total number of chunks is:

⁠N = ⌈ S / C ⌉⁠

The client first computes this $N$,n sends a request containing a unique file identifier to server asking “which chunks have already been uploaded?”. The server looks up its records and returns a list of completed chunk indices. The client resumes from next missing chunk – this is most primitive yet universally applicable resumable‑upload scheme.

至于下载方向，HTTP Range 请求

The download side relies on standard HTTP Range header. A client sends:

# Example request header
再看Range，bytes=1048576-

The server detects this header,seeks to requested offset,returns only that slice with status ` header that tells client which byte range is being transmitted. Tools like browsers,wget ‑c or curl ‑C - all use this mechanism to resume interrupted downloads.

完整案例这方面，FastAPI + MinIO 实现分片上传/下载与断点续传

The following demo uses MinIO as storage backend and SQLite to persist upload‑session state. In production you can replace SQLite with PostgreSQL or Redis without touching any business logic.

项目结构概览 

big_file_service/
├── main.py # FastAPI入口
├── storage.py # 存储后端封装
├── models.py # 分片状态数据模型
├── upload.py # 上传相关接口
├── download.py # 下载相关接口
└── requirements.txt

存储后端封装

# storage.py
from minio import Minio
from minio.error import S3Error
import os
from io import BytesIO
class ObjectStorage:
def __init__:
self.client = Minio(
"localhost:9000"。access_key="minioadmin",secret_key="minioadmin",secure=False,)
self.bucket = "big-files"
if not self.client.bucket_exists:
self.client.make_bucket
def save_chunk:
"""保存单个分片"""
object_name = f"chunks/{upload_id}/{chunk_index}"
self.client.put_object(
self.bucket,object_name,BytesIO,length=len,)
def merge_chunks:
"""使用 S3 的 compose_object 合并所有分片"""
from minio.commonconfig import ComposeSource
sources =
self.client.compose_object
# 清理临时碎片
for i in range:
self.client.remove_object
def get_object_stream:
return self.client.get_object(
self.bucket,object_name,offset=offset,length=length,)
def stat_object:
return self.client.stat_object
storage = ObjectStorage

分片状态记录

# models.py
import sqlite3
conn = sqlite3.connect
conn.execute("""
CREATE TABLE IF NOT EXISTS upload_sessions (
upload_id TEXT PRIMARY KEY,filename TEXT,file_hash TEXT,total_size INTEGER,chunk_size INTEGER,total_chunks INTEGER,uploaded_chunks TEXT DEFAULT '',status TEXT DEFAULT 'uploading'
)
""")
conn.commit
def create_session(upload_id,filename,file_hash,total_size,chunk_size,total_chunks):
conn.execute(
"INSERT OR IGNORE INTO upload_sessions VALUES ",(upload_id,filename,file_hash,total_size,chunk_size,total_chunks,'','uploading')
)
conn.commit
def get_session_by_hash:
row = conn.execute(
"SELECT * FROM upload_sessions WHERE file_hash=?",).fetchone
return row
def mark_chunk_done:
row = conn.execute(
"SELECT uploaded_chunks FROM upload_sessions WHERE upload_id=?",).fetchone
done_set = set) if row else set
done_set.add)
conn.execute(
"UPDATE upload_sessions SET uploaded_chunks=?WHERE upload_id=?",upload_id)
)
conn.commit
def get_uploaded_chunks:
row = conn.execute(
"SELECT uploaded_chunks FROM upload_sessions WHERE upload_id=?",).fetchone
if not row or not row:
return set
return set for x in row.split)

# upload.py
from fastapi import APIRouter,UploadFile,Form
from storage import storage
from models import (
create_session。get_session_by_hash,mark_chunk_done,get_uploaded_chunks,)
import uuid,math
router = APIRouter
@router.post
async def init_upload(
filename: str = Form,file_hash: str = Form,total_size: int = Form,chunk_size: int = Form
):
"""
初始化上传。如果同一 hash 已存在则返回已完成的块列表，实现“秒级”续传。"""
existing = get_session_by_hash
if existing:
upload_id = existing
uploaded = get_uploaded_chunks
return {
"upload_id": upload_id,"uploaded_chunks": sorted),"resumed": True,}
upload_id = str)
total_chunks = math.ceil
create_session(upload_id,filename,file_hash。total_size,chunk_size,total_chunks)
return {
"upload_id": upload_id,"uploaded_chunks":,"resumed": False,"total_chunks": total_chunks,}
@router.post
async def upload_chunk(
upload_id: str = Form,chunk_index: int = Form,file: UploadFile | None = None):
"""
单块上传。怎么说呢，若该块已经存在直接返回，可安全重复请求。"""
uploaded = get_uploaded_chunks
if chunk_index in uploaded:
return {"status":"already_uploaded","chunk_index":chunk_index}
data = await file.read
storage.save_chunk。data)
mark_chunk_done)
return {"status":"ok","chunk_index":chunk_index}
@router.post
async def complete_upload(
upload_id: str=Form,filename:str=Form,total_chunks:int=Form):
"""
所有块成功后调用合并。"""
storage.merge_chunks
return {"status":"completed","filename":filename}

客户端配合流程简述：

• A) 对整文件计算 SHA‑256。
• B) 调用 /upload/init with file_hash、totalSize、chunkSize.
• C) 若返回 "resumed":true，则读取已完成块列表；否则从第 0 块开始，
• D) 并发上传缺失块至 /upload/chunk。
• E) 所有块成功后调用 /upload/complete 完成合并。
• If network drops or page refreshes at any point you can repeat step B – server will remember which chunks are already stored.
\end{ul}

下载接口 – 支持 Range 请求

from fastapi import APIRouter,Request,HTTPException
from fastapi.responses import StreamingResponse
from storage import storage
router=APIRouter
@router.get
async def download_file:
stat=storage.stat_object
file_size=stat.size
 range_header=request.headers.get
if range_header:
# 示例："bytes=1048576-"
至于try，_,byte_range=range_header.split
start_str,end_str=byte_range.split
start=int if start_str else 0
end=int if end_str else file_size-1
except Exception:
raise HTTPException
length=end-start+1
stream=storage.get_object_stream
async def iter_stream:
for part in stream.stream: # 1MiB per iteration
yield part
stream.close
headers={
"Content-Range":f"bytes {start}-{end}/{file_size}"。"Accept-Ranges":"bytes","Content-Length":str,}
return StreamingResponse,status_code=206,headers=headers)
# 没有 Range → 完整下载
full_stream=storage.get_object_stream
async def iter_full:
for part in full_stream.stream:
yield part
full_stream.close
headers={"Content-Length":str,"Accept-Ranges":"bytes"}
return StreamingResponse,headers=headers)


`\



挂载路由 & 部署要点 
python
# main.py
from fastapi import FastAPI
from upload import router as upload_router
from download import router as download_router
app=FastAPI
app.include_router
app.include_router



Nginx / Traefik 等反向代理限制：
将 `clientmaxbody_size` 调高，否则大文件请求会在网关层被截断，与 FastAPI 本身无关。<\/li>

{timeout} 设置：
单个分块请求通常毫秒级。但完整合并或跨区域复制可能耗时较长，需要相应放宽网关和 Uvicorn 的超时配置。<\/li>

{并发控制}：
推荐一次性开启 ~5~10 条并发连接进行分块上传；过多会导致对象存储限流或服务器句柄耗尽。<\/li>

{清理策略}：
对于因网络异常导致残留未完成的碎片，可使用定时任务扫描 `chunks/*/*` 前缀并删除超过阈值时间 的孤儿对象。<\/li>

{如果不想自己实现协议}：直接部署 tus 服务端，FastAPI 可仅负责鉴权与元数据管理。<\/li>
<\/ul>

几条路线该怎么选？\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b\u200b‑",...