Python使用FastAPI實(shí)現(xiàn)大文件分片上傳與斷點(diǎn)續(xù)傳功能

更新時(shí)間：2025年09月12日 10:12:09 作者：楓葉V

大文件直傳常遇到超時(shí)、網(wǎng)絡(luò)抖動(dòng)失敗、失敗后只能重傳的問題,分片上傳 + 斷點(diǎn)續(xù)傳可以把大文件拆成若干小塊逐個(gè)上傳,并在中斷后從已完成分片繼續(xù),所以本文提供一套可直接運(yùn)行的 Python（FastAPI）服務(wù)端實(shí)現(xiàn),需要的朋友可以參考下

大文件直傳常遇到超時(shí)、網(wǎng)絡(luò)抖動(dòng)失敗、失敗后只能重傳的問題。分片上傳 + 斷點(diǎn)續(xù)傳可以把大文件拆成若干小塊逐個(gè)上傳，并在中斷后從已完成分片繼續(xù)，從而顯著提升成功率與用戶體驗(yàn)。

本文提供一套 可直接運(yùn)行 的 Python（FastAPI）服務(wù)端實(shí)現(xiàn)，配套 Python 客戶端腳本 與少量前端思路，支持：

分片上傳
斷點(diǎn)續(xù)傳（查詢已傳分片）
全量合并
可選內(nèi)容校驗(yàn)（SHA256/MD5）
并發(fā)安全（文件級鎖）
可平滑接入對象存儲(chǔ)（MinIO/S3）示例

一、接口設(shè)計(jì)

接口	方法	描述
/upload/init	POST	初始化上傳，創(chuàng)建 upload_id 與分片元數(shù)據(jù)
/upload/chunk	POST	上傳單個(gè)分片，參數(shù)：upload_id、index
/upload/status	GET	查詢某 upload_id 已上傳分片列表
/upload/merge	POST	全量合并分片，生成最終文件

說明：index 從 1 開始；建議固定 chunk_size（如 5MB/10MB），便于計(jì)算總分片數(shù) total_chunks = ceil(size/chunk_size)。

二、服務(wù)端實(shí)現(xiàn)（FastAPI）

2.1 運(yùn)行環(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 依賴；如需對象存儲(chǔ)支持，再安裝 minio 或 boto3。

2.2 目錄結(jié)構(gòu)建議

.
├─ server.py              # FastAPI 服務(wù)端
├─ upload.db              # SQLite 元數(shù)據(jù)庫（運(yùn)行后生成）
└─ uploads/
   └─ {upload_id}/
      ├─ chunks/         # 分片臨時(shí)目錄
      ├─ meta.json       # 元信息（冗余存儲(chǔ)，輔助排查）
      └─ final/          # 最終文件目錄

2.3 server.py（可直接運(yùn)行）

# 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  # 可選：合并后對整文件校驗(yàn)

# ---------- 路由 ----------
@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)容校驗(yàn)（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"

    # 將分片寫入臨時(shí)文件
    data = await file.read()
    if content_hash:
        # 這里兼容 md5/sha256：長度32大概率md5，長度64大概率sha256，也可自定義一個(gè) 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 校驗(yàn)失敗")

    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

            # 校驗(yàn)分片是否齊全
            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())

        # 可選：合并后哈希校驗(yàn)
        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, "合并后文件校驗(yàn)失敗")

        # 也可以在此處清理分片文件，或異步清理
        return JSONResponse({"message": "merged", "final_path": str(final_path)})
        
# ---------- 啟動(dòng) ----------
if __name__ == "__main__":
    init_db()
    uvicorn.run(app, host="0.0.0.0", port=8000)

啟動(dòng)：python server.py，訪問 http://localhost:8000/docs 可用 Swagger 調(diào)試。

三、Python 客戶端腳本（支持?jǐn)帱c(diǎn)續(xù)傳與失敗重試）

3.1 客戶端依賴

pip install requests tqdm

3.2 client.py（可直接運(yùn)行）

# 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  # 合并后校驗(yàn)整文件
    })
    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)對整文件計(jì)算 SHA256，分片使用 MD5（use_md5_for_chunk=True）進(jìn)行輕量校驗(yàn)。
可按需改為并發(fā)上傳（例如 concurrent.futures.ThreadPoolExecutor），但要關(guān)注帶寬/并發(fā)上限與服務(wù)端限流策略。

四、斷點(diǎn)續(xù)傳原理與實(shí)現(xiàn)要點(diǎn)

唯一標(biāo)識：upload_id 由 (filename, size, chunk_size, file_hash) 派生，確保同一文件同配置產(chǎn)生穩(wěn)定 ID。
狀態(tài)存儲(chǔ)：用 SQLite 記錄 uploads 與 chunks，服務(wù)重啟后仍可恢復(fù)。
狀態(tài)查詢：/upload/status 返回已上傳分片列表，客戶端跳過已完成分片，直接續(xù)傳剩余分片。
合并鎖：為避免并發(fā)重復(fù)合并，對 upload_id 加進(jìn)程內(nèi)鎖。
數(shù)據(jù)校驗(yàn)：
分片級：content_hash（MD5/SHA256）保障每塊完整。
整體級：expected_hash 合并后校驗(yàn)整文件。
失敗重試：客戶端對分片上傳進(jìn)行有限次重試，并可按需指數(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ā)上傳、進(jìn)度條展示、速度/剩余時(shí)間估算、取消/暫停/繼續(xù)、切片校驗(yàn)、錯(cuò)誤處理等。

六、接入對象存儲(chǔ)（MinIO/S3）思路

直傳對象存儲(chǔ)：客戶端直傳對象存儲(chǔ)（預(yù)簽名 URL），服務(wù)端僅做 init/status/merge 與校驗(yàn)，減少后端帶寬占用。
后端合并：分片先上傳至桶的 /{upload_id}/chunks/{index}.part，合并階段服務(wù)端從對象存儲(chǔ)流式讀出到最終對象。
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) 為每個(gè)分片生成預(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：便于人類理解與日志排查，同時(shí)避免 0 與未設(shè)置的歧義。

Q2：如何并發(fā)上傳？ A：客戶端開啟多個(gè)線程協(xié)程并發(fā)上傳，但注意限流、連接上限、后端寫盤壓力。通常 3~8 并發(fā)較穩(wěn)妥。

Q3：如何“秒傳”？ A：在 /upload/init 前，客戶端先計(jì)算整文件 hash（如 SHA256），服務(wù)端判斷是否已有該文件，若存在則直接返回完成態(tài)。

Q4：如何避免“同名不同內(nèi)容”沖突？ A：upload_id 的推導(dǎo)包含 size/chunk_size/file_hash，同名文件內(nèi)容不同會(huì)產(chǎn)生不同 upload_id，可兼容。

Q5：如何防止惡意超大文件或分片風(fēng)暴？ A：加入鑒權(quán)、白名單、配額、單 IP 并發(fā)限制、請求速率限制、最大分片大小及總大小限制。

八、總結(jié)

本文給出 Python（FastAPI）實(shí)現(xiàn)的大文件分片上傳與斷點(diǎn)續(xù)傳的 可運(yùn)行模板，同時(shí)提供了 客戶端腳本 與 前端思路。在生產(chǎn)環(huán)境，建議疊加：

鑒權(quán)與簽名（Token/AK/SK）
請求限流與并發(fā)控制
存儲(chǔ)與計(jì)算解耦（對象存儲(chǔ)直傳/多段上傳）
完整的監(jiān)控告警與清理策略
更完備的異?；謴?fù)與審計(jì)日志

你可以以此為基線，快速擴(kuò)展到 MinIO/S3 等對象存儲(chǔ)，并對接你現(xiàn)有的網(wǎng)關(guān)與權(quán)限體系。

以上就是Python使用FastAPI實(shí)現(xiàn)大文件分片上傳與斷點(diǎn)續(xù)傳功能的詳細(xì)內(nèi)容，更多關(guān)于Python FastAPI分片上傳與斷點(diǎn)續(xù)傳的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章:

国产无遮挡裸体免费直播视频,久久精品国产蜜臀av,动漫在线视频一区二区,欧亚日韩一区二区三区,久艹在线免费视频,国产精品美女网站免费,正在播放 97超级视频在线观看,斗破苍穹年番在线观看免费,51最新乱码中文字幕

Python使用FastAPI實(shí)現(xiàn)大文件分片上傳與斷點(diǎn)續(xù)傳功能

目錄

一、接口設(shè)計(jì)

二、服務(wù)端實(shí)現(xiàn)（FastAPI）

2.1 運(yùn)行環(huán)境

2.2 目錄結(jié)構(gòu)建議

2.3 server.py（可直接運(yùn)行）

三、Python 客戶端腳本（支持?jǐn)帱c(diǎn)續(xù)傳與失敗重試）

3.1 客戶端依賴

3.2 client.py（可直接運(yùn)行）

四、斷點(diǎn)續(xù)傳原理與實(shí)現(xiàn)要點(diǎn)

五、前端思路（可選）

六、接入對象存儲(chǔ)（MinIO/S3）思路

七、常見問題

八、總結(jié)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

国产无遮挡裸体免费直播视频,久久精品国产蜜臀av,动漫在线视频一区二区,欧亚日韩一区二区三区,久艹在线 免费视频,国产精品美女网站免费,正在播放 97超级视频在线观看,斗破苍穹年番在线观看免费,51最新乱码中文字幕

Python使用FastAPI實(shí)現(xiàn)大文件分片上傳與斷點(diǎn)續(xù)傳功能

目錄

一、接口設(shè)計(jì)

二、服務(wù)端實(shí)現(xiàn)（FastAPI）

2.1 運(yùn)行環(huán)境

2.2 目錄結(jié)構(gòu)建議

2.3 server.py（可直接運(yùn)行）

三、Python 客戶端腳本（支持?jǐn)帱c(diǎn)續(xù)傳與失敗重試）

3.1 客戶端依賴

3.2 client.py（可直接運(yùn)行）

四、斷點(diǎn)續(xù)傳原理與實(shí)現(xiàn)要點(diǎn)

五、前端思路（可選）

六、接入對象存儲(chǔ)（MinIO/S3）思路

七、常見問題

八、總結(jié)

相關(guān)文章

最新評論

大家感興趣的內(nèi)容

最近更新的內(nèi)容

常用在線小工具

国产无遮挡裸体免费直播视频,久久精品国产蜜臀av,动漫在线视频一区二区,欧亚日韩一区二区三区,久艹在线免费视频,国产精品美女网站免费,正在播放 97超级视频在线观看,斗破苍穹年番在线观看免费,51最新乱码中文字幕

一、接口設(shè)計(jì)

二、服務(wù)端實(shí)現(xiàn)（FastAPI）

三、Python 客戶端腳本（支持?jǐn)帱c(diǎn)續(xù)傳與失敗重試）

五、前端思路（可選）

六、接入對象存儲(chǔ)（MinIO/S3）思路

八、總結(jié)