HTTPX文件上传支持分片吗

wen python案例 2026-07-11 1

本文目录导读：

HTTPX文件上传支持分片吗

HTTPX 原生支持情况
如何用 HTTPX 实现分片上传
更完善的实现（带重试和进度）
服务端配合
替代方案

HTTPX 是否支持分片上传，答案是：HTTPX 本身不内置分片上传功能,但可以通过手动实现来支持。

更详细的分析如下：

HTTPX 原生支持情况

HTTPX 是一个通用的 HTTP 客户端库,它提供了：

基本的文件上传（files 参数）
流式上传（content 参数）
多部分表单上传

但它没有像某些专门的文件上传组件（如 django-chunked-upload、tus 等）那样内置的分片上传逻辑。

如何用 HTTPX 实现分片上传

你需要手动实现分片上传逻辑,基本原理是：

import httpx
import os
def upload_file_in_chunks(file_path, upload_url, chunk_size=5*1024*1024):
    """
    分片上传文件
    :param file_path: 文件路径
    :param upload_url: 上传接口地址
    :param chunk_size: 每片大小（字节），默认5MB
    """
    file_size = os.path.getsize(file_path)
    total_chunks = (file_size + chunk_size - 1) // chunk_size
    with open(file_path, 'rb') as f:
        for chunk_index in range(total_chunks):
            # 读取当前分片
            chunk_data = f.read(chunk_size)
            # 准备分片元数据
            files = {
                'chunk': ('chunk', chunk_data, 'application/octet-stream'),
                'chunk_index': (None, str(chunk_index)),
                'total_chunks': (None, str(total_chunks)),
                'filename': (None, os.path.basename(file_path))
            }
            # 上传分片
            with httpx.Client() as client:
                response = client.post(upload_url, files=files)
            if response.status_code != 200:
                print(f"分片 {chunk_index + 1}/{total_chunks} 上传失败")
                return False
            print(f"分片 {chunk_index + 1}/{total_chunks} 上传成功")
    return True
# 使用示例
upload_file_in_chunks('large_file.zip', 'https://api.example.com/upload')

更完善的实现（带重试和进度）

import httpx
import os
from typing import Optional
class ChunkedUploader:
    def __init__(self, upload_url: str, chunk_size: int = 5*1024*1024, max_retries: int = 3):
        self.upload_url = upload_url
        self.chunk_size = chunk_size
        self.max_retries = max_retries
    def upload(self, file_path: str) -> bool:
        file_size = os.path.getsize(file_path)
        total_chunks = (file_size + self.chunk_size - 1) // self.chunk_size
        with open(file_path, 'rb') as f:
            for chunk_index in range(total_chunks):
                chunk_data = f.read(self.chunk_size)
                # 带重试的上传
                success = False
                for attempt in range(self.max_retries):
                    try:
                        with httpx.Client(timeout=30.0) as client:
                            response = client.post(
                                self.upload_url,
                                files={
                                    'chunk': ('chunk', chunk_data),
                                    'chunk_index': (None, str(chunk_index)),
                                    'total_chunks': (None, str(total_chunks)),
                                    'filename': (None, os.path.basename(file_path))
                                }
                            )
                        if response.status_code == 200:
                            success = True
                            # 显示进度
                            progress = (chunk_index + 1) / total_chunks * 100
                            print(f"进度: {progress:.1f}%")
                            break
                        else:
                            print(f"尝试 {attempt + 1} 失败，状态码: {response.status_code}")
                    except httpx.RequestError as e:
                        print(f"尝试 {attempt + 1} 失败: {str(e)}")
                if not success:
                    print(f"分片 {chunk_index + 1} 上传失败，终止上传")
                    return False
        print("文件上传完成")
        return True