Python使用FastAPI實現(xiàn)大文件分片上傳與斷點續(xù)傳功能
更新時間：2025年09月12日 10:12:09 作者：楓葉V
大文件直傳常遇到超時、網(wǎng)絡(luò)抖動失敗、失敗后只能重傳的問題,分片上傳 + 斷點續(xù)傳可以把大文件拆成若干小塊逐個上傳,并在中斷后從已完成分片繼續(xù),所以本文提供一套 可直接運行 的 Python（FastAPI）服務(wù)端實現(xiàn),需要的朋友可以參考下
大文件直傳常遇到超時、網(wǎng)絡(luò)抖動失敗、失敗后只能重傳的問題。分片上傳 + 斷點續(xù)傳可以把大文件拆成若干小塊逐個上傳，并在中斷后從已完成分片繼續(xù)，從而顯著提升成功率與用戶體驗。
本文提供一套 可直接運行 的 Python（FastAPI）服務(wù)端實現(xiàn)，配套 Python 客戶端腳本 與少量前端思路，支持：
- 分片上傳
- 斷點續(xù)傳（查詢已傳分片）
- 全量合并
- 可選內(nèi)容校驗（SHA256/MD5）
- 并發(fā)安全（文件級鎖）
- 可平滑接入對象存儲（MinIO/S3）示例
一、接口設(shè)計
| 接口 | 方法 | 描述 |
|---|---|---|
| /upload/init | POST | 初始化上傳，創(chuàng)建 upload_id 與分片元數(shù)據(jù) |
| /upload/chunk | POST | 上傳單個分片，參數(shù)：upload_id、index |
| /upload/status | GET | 查詢某 upload_id 已上傳分片列表 |
| /upload/merge | POST | 全量合并分片，生成最終文件 |
說明：index 從 1 開始；建議固定 chunk_size（如 5MB/10MB），便于計算總分片數(shù) total_chunks = ceil(size/chunk_size)。
二、服務(wù)端實現(xiàn)（FastAPI）
2.1 運行環(huán)境
python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate pip install fastapi uvicorn pydantic[dotenv] python-multipart
注：本文僅用到標(biāo)準(zhǔn)庫 + FastAPI 依賴；如需對象存儲支持，再安裝 minio 或 boto3。
2.2 目錄結(jié)構(gòu)建議
.
├─ server.py # FastAPI 服務(wù)端
├─ upload.db # SQLite 元數(shù)據(jù)庫（運行后生成）
└─ uploads/
└─ {upload_id}/
├─ chunks/ # 分片臨時目錄
├─ meta.json # 元信息（冗余存儲，輔助排查）
└─ final/ # 最終文件目錄
2.3 server.py（可直接運行）
# server.py
import hashlib
import json
import os
import sqlite3
import threading
from contextlib import contextmanager
from math import ceil
from pathlib import Path
from typing import List, Optional
from fastapi import FastAPI, UploadFile, File, HTTPException, Query, Form
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import uvicorn
BASE_DIR = Path(__file__).parent.resolve()
DATA_DIR = BASE_DIR / "uploads"
DB_PATH = BASE_DIR / "upload.db"
DATA_DIR.mkdir(exist_ok=True)
app = FastAPI(title="Chunked Upload (FastAPI)")
# ---------- 并發(fā)鎖（同一 upload_id 合并上鎖） ----------
_merge_locks = {}
_merge_locks_global = threading.Lock()
def _get_lock(upload_id: str) -> threading.Lock:
with _merge_locks_global:
if upload_id not in _merge_locks:
_merge_locks[upload_id] = threading.Lock()
return _merge_locks[upload_id]
# ---------- SQLite 簡單封裝 ----------
def init_db():
with sqlite3.connect(DB_PATH) as conn:
c = conn.cursor()
c.execute("""CREATE TABLE IF NOT EXISTS uploads(
upload_id TEXT PRIMARY KEY,
filename TEXT,
size INTEGER,
chunk_size INTEGER,
total_chunks INTEGER,
file_hash TEXT
)""")
c.execute("""CREATE TABLE IF NOT EXISTS chunks(
upload_id TEXT,
idx INTEGER,
size INTEGER,
hash TEXT,
PRIMARY KEY(upload_id, idx)
)""")
conn.commit()
@contextmanager
def db() :
conn = sqlite3.connect(DB_PATH)
try:
yield conn
finally:
conn.close()
# ---------- 工具 ----------
def sha256_of_bytes(b: bytes) -> str:
h = hashlib.sha256()
h.update(b)
return h.hexdigest()
def md5_of_bytes(b: bytes) -> str:
h = hashlib.md5()
h.update(b)
return h.hexdigest()
def ensure_upload_folders(upload_id: str) -> Path:
root = DATA_DIR / upload_id
(root / "chunks").mkdir(parents=True, exist_ok=True)
(root / "final").mkdir(parents=True, exist_ok=True)
return root
def write_meta(root: Path, meta: dict):
(root / "meta.json").write_text(json.dumps(meta, ensure_ascii=False, indent=2))
# ---------- 請求/響應(yīng)模型 ----------
class InitReq(BaseModel):
filename: str
size: int
chunk_size: int
file_hash: Optional[str] = None # 可選：客戶端傳文件整體 SHA256/MD5
class InitResp(BaseModel):
upload_id: str
total_chunks: int
class StatusResp(BaseModel):
upload_id: str
uploaded: List[int]
total_chunks: int
class MergeReq(BaseModel):
upload_id: str
expected_hash: Optional[str] = None # 可選：合并后對整文件校驗
# ---------- 路由 ----------
@app.post("/upload/init", response_model=InitResp)
def init_upload(req: InitReq):
if req.size <= 0 or req.chunk_size <= 0:
raise HTTPException(400, "size/chunk_size 必須為正整數(shù)")
total = ceil(req.size / req.chunk_size)
upload_id = hashlib.sha1(f"{req.filename}:{req.size}:{req.chunk_size}:{req.file_hash or ''}".encode()).hexdigest()
root = ensure_upload_folders(upload_id)
with db() as conn:
c = conn.cursor()
# UPSERT
c.execute("INSERT OR IGNORE INTO uploads(upload_id, filename, size, chunk_size, total_chunks, file_hash) VALUES(?,?,?,?,?,?)",
(upload_id, req.filename, req.size, req.chunk_size, total, req.file_hash))
conn.commit()
write_meta(root, {
"upload_id": upload_id,
"filename": req.filename,
"size": req.size,
"chunk_size": req.chunk_size,
"total_chunks": total,
"file_hash": req.file_hash
})
return InitResp(upload_id=upload_id, total_chunks=total)
@app.post("/upload/chunk")
async def upload_chunk(
upload_id: str = Query(..., description="init 返回的 upload_id"),
index: int = Query(..., ge=1, description="分片序號，從1開始"),
content_hash: Optional[str] = Query(None, description="可選：分片內(nèi)容校驗（sha256或md5，客戶端自定義）"),
file: UploadFile = File(...)
):
# 檢查 upload 是否存在
with db() as conn:
c = conn.cursor()
row = c.execute("SELECT total_chunks FROM uploads WHERE upload_id=?", (upload_id,)).fetchone()
if not row:
raise HTTPException(404, "upload_id 不存在，請先 /upload/init")
total_chunks = row[0]
if index > total_chunks:
raise HTTPException(400, f"index 超出總分片數(shù)：{total_chunks}")
root = ensure_upload_folders(upload_id)
chunk_path = root / "chunks" / f"{index}.part"
# 將分片寫入臨時文件
data = await file.read()
if content_hash:
# 這里兼容 md5/sha256：長度32大概率md5，長度64大概率sha256，也可自定義一個 query param 指定算法
calc = md5_of_bytes(data) if len(content_hash) == 32 else sha256_of_bytes(data)
if calc != content_hash:
raise HTTPException(400, "content_hash 校驗失敗")
with open(chunk_path, "wb") as f:
f.write(data)
# 記錄 DB
with db() as conn:
c = conn.cursor()
c.execute("INSERT OR REPLACE INTO chunks(upload_id, idx, size, hash) VALUES(?,?,?,?)",
(upload_id, index, len(data), content_hash))
conn.commit()
return JSONResponse({"message": "ok", "index": index, "size": len(data)})
@app.get("/upload/status", response_model=StatusResp)
def upload_status(upload_id: str):
with db() as conn:
c = conn.cursor()
up = c.execute("SELECT total_chunks FROM uploads WHERE upload_id=?", (upload_id,)).fetchone()
if not up:
raise HTTPException(404, "upload_id 不存在")
total = up[0]
rows = c.execute("SELECT idx FROM chunks WHERE upload_id=? ORDER BY idx", (upload_id,)).fetchall()
uploaded = [r[0] for r in rows]
return StatusResp(upload_id=upload_id, uploaded=uploaded, total_chunks=total)
@app.post("/upload/merge")
def upload_merge(req: MergeReq):
# 防止并發(fā)重復(fù)合并
lock = _get_lock(req.upload_id)
with lock:
with db() as conn:
c = conn.cursor()
up = c.execute("SELECT filename, total_chunks FROM uploads WHERE upload_id=?", (req.upload_id,)).fetchone()
if not up:
raise HTTPException(404, "upload_id 不存在")
filename, total = up
# 校驗分片是否齊全
rows = c.execute("SELECT idx FROM chunks WHERE upload_id=? ORDER BY idx", (req.upload_id,)).fetchall()
uploaded = [r[0] for r in rows]
if len(uploaded) != total or uploaded != list(range(1, total + 1)):
raise HTTPException(400, f"分片不完整，已上傳：{uploaded}，應(yīng)為 1..{total}")
root = ensure_upload_folders(req.upload_id)
final_path = root / "final" / filename
# 合并
with open(final_path, "wb") as dst:
for i in range(1, total + 1):
chunk_path = root / "chunks" / f"{i}.part"
with open(chunk_path, "rb") as src:
dst.write(src.read())
# 可選：合并后哈希校驗
if req.expected_hash:
with open(final_path, "rb") as f:
data = f.read()
calc = md5_of_bytes(data) if len(req.expected_hash) == 32 else sha256_of_bytes(data)
if calc != req.expected_hash:
raise HTTPException(400, "合并后文件校驗失敗")
# 也可以在此處清理分片文件，或異步清理
return JSONResponse({"message": "merged", "final_path": str(final_path)})
# ---------- 啟動 ----------
if __name__ == "__main__":
init_db()
uvicorn.run(app, host="0.0.0.0", port=8000)
啟動：python server.py，訪問 http://localhost:8000/docs 可用 Swagger 調(diào)試。
三、Python 客戶端腳本（支持?jǐn)帱c續(xù)傳與失敗重試）
3.1 客戶端依賴
pip install requests tqdm
3.2 client.py（可直接運行）
# client.py
import math
import os
import time
import hashlib
import requests
from pathlib import Path
from tqdm import tqdm
SERVER = "http://localhost:8000"
def sha256_of_file(path: Path) -> str:
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(1024 * 1024), b""):
h.update(chunk)
return h.hexdigest()
def md5_of_bytes(b: bytes) -> str:
import hashlib
h = hashlib.md5()
h.update(b)
return h.hexdigest()
def upload_file(path: str, chunk_size: int = 5 * 1024 * 1024, use_md5_for_chunk=False, retries=3, sleep=1):
p = Path(path)
size = p.stat().st_size
file_hash = sha256_of_file(p) # 也可改為 MD5
total_chunks = math.ceil(size / chunk_size)
# 1) init
r = requests.post(f"{SERVER}/upload/init", json={
"filename": p.name,
"size": size,
"chunk_size": chunk_size,
"file_hash": file_hash
})
r.raise_for_status()
j = r.json()
upload_id = j["upload_id"]
total_chunks = j["total_chunks"]
print("upload_id:", upload_id, "total_chunks:", total_chunks)
# 2) 查詢已上傳
def get_status():
rr = requests.get(f"{SERVER}/upload/status", params={"upload_id": upload_id})
rr.raise_for_status()
return rr.json()["uploaded"]
uploaded = set(get_status())
# 3) 逐片上傳（可按需并發(fā)）
with open(p, "rb") as f:
for index in tqdm(range(1, total_chunks + 1), desc="Uploading"):
if index in uploaded:
continue
start = (index - 1) * chunk_size
end = min(size, index * chunk_size)
f.seek(start)
data = f.read(end - start)
content_hash = md5_of_bytes(data) if use_md5_for_chunk else None
for attempt in range(retries):
try:
files = {"file": (f"{index}.part", data, "application/octet-stream")}
params = {"upload_id": upload_id, "index": index}
if content_hash:
params["content_hash"] = content_hash
rr = requests.post(f"{SERVER}/upload/chunk", files=files, params=params, timeout=30)
rr.raise_for_status()
break
except Exception as e:
print(f"[{index}] retry {attempt+1}/{retries} because {e}")
time.sleep(sleep)
else:
raise RuntimeError(f"upload chunk {index} failed after retries")
# 4) 合并
r = requests.post(f"{SERVER}/upload/merge", json={
"upload_id": upload_id,
"expected_hash": file_hash # 合并后校驗整文件
})
r.raise_for_status()
print("merge result:", r.json())
if __name__ == "__main__":
# 示例：python client.py
upload_file("bigfile.bin", chunk_size=5 * 1024 * 1024, use_md5_for_chunk=True)
說明：
- 默認(rèn)對整文件計算 SHA256，分片使用 MD5（use_md5_for_chunk=True）進行輕量校驗。
- 可按需改為并發(fā)上傳（例如 concurrent.futures.ThreadPoolExecutor），但要關(guān)注帶寬/并發(fā)上限與服務(wù)端限流策略。
四、斷點續(xù)傳原理與實現(xiàn)要點
- 唯一標(biāo)識：upload_id 由 (filename, size, chunk_size, file_hash) 派生，確保同一文件同配置產(chǎn)生穩(wěn)定 ID。
- 狀態(tài)存儲：用 SQLite 記錄 uploads 與 chunks，服務(wù)重啟后仍可恢復(fù)。
- 狀態(tài)查詢：/upload/status 返回已上傳分片列表，客戶端跳過已完成分片，直接續(xù)傳剩余分片。
- 合并鎖：為避免并發(fā)重復(fù)合并，對 upload_id 加進程內(nèi)鎖。
- 數(shù)據(jù)校驗：
- 分片級：content_hash（MD5/SHA256）保障每塊完整。
- 整體級：expected_hash 合并后校驗整文件。
- 失敗重試：客戶端對分片上傳進行有限次重試，并可按需指數(shù)回退。
- 清理策略：合并成功后可異步清理分片文件，釋放磁盤。
五、前端思路（可選）
Web 前端（瀏覽器）常用 Blob.slice 分片 + fetch/XMLHttpRequest 上傳。示例片段：
async function upload(file) {
const chunkSize = 5 * 1024 * 1024;
const totalChunks = Math.ceil(file.size / chunkSize);
// 1) init
const initResp = await fetch('/upload/init', {
method: 'POST',
headers: { 'Content-Type': 'application/json'},
body: JSON.stringify({ filename: file.name, size: file.size, chunk_size: chunkSize })
}).then(r => r.json());
const { upload_id } = initResp;
// 2) 查詢已上傳
const status = await fetch(`/upload/status?upload_id=${upload_id}`).then(r => r.json());
const uploaded = new Set(status.uploaded);
// 3) 順序或并發(fā)上傳
for (let i = 1; i <= totalChunks; i++) {
if (uploaded.has(i)) continue;
const start = (i - 1) * chunkSize;
const end = Math.min(file.size, i * chunkSize);
const blob = file.slice(start, end);
const form = new FormData();
form.append('file', blob, `${i}.part`);
await fetch(`/upload/chunk?upload_id=${upload_id}&index=${i}`, { method: 'POST', body: form });
}
// 4) merge
await fetch('/upload/merge', {
method: 'POST',
headers: { 'Content-Type': 'application/json'},
body: JSON.stringify({ upload_id })
});
}
生產(chǎn)環(huán)境可加入：并發(fā)上傳、進度條展示、速度/剩余時間估算、取消/暫停/繼續(xù)、切片校驗、錯誤處理等。
六、接入對象存儲（MinIO/S3）思路
- 直傳對象存儲：客戶端直傳對象存儲（預(yù)簽名 URL），服務(wù)端僅做 init/status/merge 與校驗，減少后端帶寬占用。
- 后端合并：分片先上傳至桶的 /{upload_id}/chunks/{index}.part，合并階段服務(wù)端從對象存儲流式讀出到最終對象。
- S3 多段上傳（Multipart Upload）：直接使用 S3 的 Multipart API，可繞過本地分片落盤，提升性能與可靠性。
示例（僅演示方向，未接入到上面代碼）：
# pip install minio
from minio import Minio
client = Minio("127.0.0.1:9000", access_key="minioadmin", secret_key="minioadmin", secure=False)
# 1) 為每個分片生成預(yù)簽名 PUT URL，前端直接 PUT 到 MinIO
url = client.presigned_put_object("my-bucket", f"{upload_id}/chunks/{index}.part", expires=timedelta(hours=1))
# 2) 合并階段：服務(wù)端從 MinIO 流式讀取所有分片，寫入最終對象
七、常見問題
Q1：為什么 index 從 1 開始？ A：便于人類理解與日志排查，同時避免 0 與未設(shè)置的歧義。
Q2：如何并發(fā)上傳？ A：客戶端開啟多個線程協(xié)程并發(fā)上傳，但注意限流、連接上限、后端寫盤壓力。通常 3~8 并發(fā)較穩(wěn)妥。
Q3：如何“秒傳”？ A：在 /upload/init 前，客戶端先計算整文件 hash（如 SHA256），服務(wù)端判斷是否已有該文件，若存在則直接返回完成態(tài)。
Q4：如何避免“同名不同內(nèi)容”沖突？ A：upload_id 的推導(dǎo)包含 size/chunk_size/file_hash，同名文件內(nèi)容不同會產(chǎn)生不同 upload_id，可兼容。
Q5：如何防止惡意超大文件或分片風(fēng)暴？ A：加入鑒權(quán)、白名單、配額、單 IP 并發(fā)限制、請求速率限制、最大分片大小及總大小限制。
八、總結(jié)
本文給出 Python（FastAPI）實現(xiàn)的大文件分片上傳與斷點續(xù)傳的 可運行模板，同時提供了 客戶端腳本 與 前端思路。在生產(chǎn)環(huán)境，建議疊加：
- 鑒權(quán)與簽名（Token/AK/SK）
- 請求限流與并發(fā)控制
- 存儲與計算解耦（對象存儲直傳/多段上傳）
- 完整的監(jiān)控告警與清理策略
- 更完備的異常恢復(fù)與審計日志
你可以以此為基線，快速擴展到 MinIO/S3 等對象存儲，并對接你現(xiàn)有的網(wǎng)關(guān)與權(quán)限體系。
以上就是Python使用FastAPI實現(xiàn)大文件分片上傳與斷點續(xù)傳功能的詳細內(nèi)容，更多關(guān)于Python FastAPI分片上傳與斷點續(xù)傳的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
在講解如何解決migrate報錯原因前，我們先要了解migrate做了什么事情，本文就詳細的介紹migrate使用以及出現(xiàn)問題的解決，感興趣的可以了解一下2021-05-05
關(guān)于Python?Selenium自動化導(dǎo)出新版WOS(web?of?science)檢索結(jié)果的問題
這篇文章主要介紹了Python?Selenium自動化導(dǎo)出新版WOS(web?of?science)檢索結(jié)果,本代碼屬于半自動化導(dǎo)出，考慮到開發(fā)效率等因素，有兩處在首次導(dǎo)出時需要手動操作，具體實現(xiàn)過程跟隨小編一起看看吧2022-01-01
本文主要為大家介紹了通過Python實現(xiàn)的免費音樂下載器，文中的示例代碼講解詳細，對我們的學(xué)習(xí)或工作有一定的幫助，需要的小伙伴可以學(xué)習(xí)一下2021-12-12
今天小編就為大家分享一篇python迭代dict的key和value的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-07-07
這篇文章主要給大家介紹了關(guān)于如何使用Python從字符串中提取數(shù)字的相關(guān)資料,我們在進行數(shù)據(jù)處理時候,可能想要提取字符串中的數(shù)字進行分析,需要的朋友可以參考下2023-11-11
在本篇文章里小編給大家整理關(guān)于python中溫度單位轉(zhuǎn)換的實例方法，有需要的朋友們可以學(xué)習(xí)參考下。2020-12-12
這篇文章主要介紹了Python中的路徑問題，幫助大家更好的理解和學(xué)習(xí)python，感興趣的朋友可以了解下2020-09-09
這篇文章主要介紹了python傳遞參數(shù)方式,實例總結(jié)了Python常用的參數(shù)傳遞方式,具有一定參考借鑒價值,需要的朋友可以參考下2015-04-04
這篇文章主要介紹了python之列表推導(dǎo)式的用法，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-11-11
Caffe卷積神經(jīng)網(wǎng)絡(luò)視覺層Vision?Layers及參數(shù)詳解
這篇文章主要為大家介紹了Caffe卷積神經(jīng)網(wǎng)絡(luò)視覺層Vision?Layers及參數(shù)詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2022-06-06