我要评论前言
日常开发中使用 python 调用 http post 接口时，如果上传报文体积较大（大量 json、批量数据、文本内容），未压缩的数据会占用更多带宽，增加网络耗时，高并发场景下还会提升服务器流量压力。
很多开发者熟悉服务端返回压缩响应（accept-encoding），却很少了解：客户端同样可以将请求 body 压缩后上传。
本文讲解 http 请求体压缩原理，提供 requests、httpx、aiohttp 完整实现代码，并梳理常见踩坑点。
关键概念区分
accept-encoding: gzip, deflate：告诉服务端，客户端支持接收压缩响应（下行压缩，绝大多数库自动支持）content-encoding: gzip：告诉服务端，当前 post 请求 body 已经被压缩（上行压缩，需要手动实现）
重要前提：服务端必须实现对应解压逻辑。若后端没有处理 content-encoding 请求头，直接接收压缩二进制流会出现解析报错。
一、实现原理
- 将原始报文（json/二进制文本）序列化为字节；
- 使用 gzip/deflate 算法压缩字节数据；
- 请求头增加
content-encoding: gzip； - post 使用
data/content参数传入压缩后的二进制，禁止使用 json 参数； - http 自动填充
content-length，无需手动计算。
推荐算法：gzip，跨语言、各后端框架兼容性最优。
二、requests 同步实现（最常用）
import requests
import gzip
import json
from io import bytesio
def gzip_compress(raw_data: bytes) -> bytes:
"""gzip压缩字节数据"""
buffer = bytesio()
with gzip.gzipfile(fileobj=buffer, mode="wb") as fp:
fp.write(raw_data)
return buffer.getvalue()
if __name__ == "__main__":
url = "http://127.0.0.1:8000/api/upload"
payload = {
"list": list(range(2000)),
"content": "大量测试文本" * 500
}
# 序列化为json字节
raw_bytes = json.dumps(payload, ensure_ascii=false).encode("utf-8")
# 阈值判断：过小的数据不压缩，避免压缩头部导致体积变大
compress_threshold = 2048
headers = {
"content-type": "application/json; charset=utf-8"
}
if len(raw_bytes) > compress_threshold:
body = gzip_compress(raw_bytes)
headers["content-encoding"] = "gzip"
else:
body = raw_bytes
resp = requests.post(url, headers=headers, data=body)
print("状态码：", resp.status_code)
print("响应内容：", resp.text)
避坑：不要使用 requests.post(..., json=payload)
json 参数内部会自动编码，无法传入压缩后的二进制数据，必须使用 data。
三、httpx 同步/异步实现（新项目推荐）
import httpx
import gzip
import json
from io import bytesio
def gzip_compress(raw_data: bytes) -> bytes:
buffer = bytesio()
with gzip.gzipfile(fileobj=buffer, mode="wb") as fp:
fp.write(raw_data)
return buffer.getvalue()
def sync_request():
url = "http://127.0.0.1:8000/api/upload"
payload = {"data": list(range(3000))}
raw_bytes = json.dumps(payload).encode("utf-8")
compressed = gzip_compress(raw_bytes)
headers = {
"content-type": "application/json; charset=utf-8",
"content-encoding": "gzip"
}
with httpx.client() as client:
r = client.post(url, headers=headers, content=compressed)
print(r.status_code, r.text)
# 异步版本
import asyncio
async def async_request():
url = "http://127.0.0.1:8000/api/upload"
payload = {"data": list(range(3000))}
raw_bytes = json.dumps(payload).encode("utf-8")
compressed = gzip_compress(raw_bytes)
headers = {
"content-type": "application/json; charset=utf-8",
"content-encoding": "gzip"
}
async with httpx.asyncclient() as client:
r = await client.post(url, headers=headers, content=compressed)
print(r.status_code, r.text)
if __name__ == "__main__":
sync_request()
# asyncio.run(async_request())
四、aiohttp 异步实现
import aiohttp
import asyncio
import gzip
import json
from io import bytesio
def gzip_compress(raw_data: bytes) -> bytes:
buffer = bytesio()
with gzip.gzipfile(fileobj=buffer, mode="wb") as fp:
fp.write(raw_data)
return buffer.getvalue()
async def main():
url = "http://127.0.0.1:8000/api/upload"
payload = {"data": list(range(3000))}
raw_bytes = json.dumps(payload).encode("utf-8")
compressed_body = gzip_compress(raw_bytes)
headers = {
"content-type": "application/json; charset=utf-8",
"content-encoding": "gzip"
}
async with aiohttp.clientsession() as session:
async with session.post(url, data=compressed_body, headers=headers) as resp:
print("status:", resp.status)
print(await resp.text())
if __name__ == "__main__":
asyncio.run(main())
五、配套服务端简易解压示例（fastapi）
测试时需要后端支持解压，附上最简示例：
from fastapi import fastapi, request
import gzip
import json
from io import bytesio
app = fastapi()
@app.post("/api/upload")
async def upload(request: request):
headers = request.headers
body = await request.body()
# 判断是否gzip压缩
if headers.get("content-encoding") == "gzip":
buf = bytesio(body)
with gzip.gzipfile(fileobj=buf) as f:
body = f.read()
data = json.loads(body)
return {"msg": "接收成功", "data": data}
六、常见问题与优化建议
1. 小数据不要强行压缩
gzip 自带固定头部（十几字节），报文很小的时候，压缩后体积可能大于原始数据。建议设置阈值（2kb左右），超过阈值再开启压缩。
2. content-type 不要删除
content-encoding 只是标记编码方式，不能替代 content-type。json 请求依旧保留 application/json。
3. 代理、网关兼容性问题
部分老旧反向代理、防火墙不识别 content-encoding，可能直接丢弃请求。上线前务必做连通性测试。
4. 区分上行压缩与下行压缩
很多人混淆两个头：
accept-encoding：客户端声明能接收压缩响应（下载）content-encoding：请求体已经压缩（上传）
requests、httpx 默认自带 accept-encoding，响应自动解压，无需手动处理。
5. 二进制文件场景
不仅限于 json，上传文本、csv 等均可使用同样逻辑；如果是图片、视频等本身已压缩的文件，再次 gzip 收益极低，不建议二次压缩。
七、适用场景总结
✅ 适合：批量上报接口、大量json文本、日志上报、大数据同步
❌ 不适合：极小报文、图片/视频等已压缩媒体文件
当接口传输量大、网络带宽有限、跨机房调用时，开启请求体压缩能有效降低传输耗时，减少流量开销。
以上就是python中http请求体压缩的实现方法的详细内容，更多关于python http请求体压缩的资料请关注代码网其它相关文章！
发表评论