Python中HTTP请求体压缩的实现方法

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)

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())

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())

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}