Python实现HTTP上行请求体的gzip压缩_Python

前言

先重申核心原理：上行请求体压缩不是http库自带功能，本质是你自己把body gzip压缩，带上 content-encoding: gzip 请求头，将压缩二进制作为请求体发送。

所有支持自定义请求body、自定义请求头的http客户端都能实现，区别只是各个库传参名称不一样：

requests：data=
httpx：content=
aiohttp：data=
urllib：data=
pycurl：postfields

下面分「同步库」「异步库」给出完整示例，统一复用gzip压缩函数。

通用压缩工具函数（所有示例共用）

import gzip
from io import bytesio

def gzip_compress(raw_bytes: bytes) -> bytes:
    buf = bytesio()
    with gzip.gzipfile(fileobj=buf, mode="wb") as f:
        f.write(raw_bytes)
    return buf.getvalue()

一、标准库：urllib（无需额外安装，内置）

无需安装第三方包，适合环境受限不能装requests的场景

import json
from urllib import request

url = "http://127.0.0.1:8000/api/upload"
payload = {"data": list(range(2000))}
raw = json.dumps(payload).encode("utf-8")
compressed_body = gzip_compress(raw)

headers = {
    "content-type": "application/json; charset=utf-8",
    "content-encoding": "gzip"
}

req = request.request(url, data=compressed_body, headers=headers, method="post")
with request.urlopen(req) as resp:
    print(resp.read().decode("utf-8"))

缺点：api简陋，不支持连接池、超时易用性差，生产一般不首选。

二、httpx（同步 + 异步，当前主流推荐）

新一代http客户端，兼容requests风格，支持http1.1/http2

import httpx
import json

url = "http://127.0.0.1:8000/api/upload"
payload = {"data": list(range(2000))}
raw = json.dumps(payload).encode("utf-8")
compressed_body = gzip_compress(raw)

headers = {
    "content-type": "application/json; charset=utf-8",
    "content-encoding": "gzip"
}

# 同步
with httpx.client(timeout=10) as client:
    res = client.post(url, headers=headers, content=compressed_body)
    print(res.text)

# 异步版本
"""
import asyncio
async def run():
    async with httpx.asyncclient(timeout=10) as client:
        res = await client.post(url, headers=headers, content=compressed_body)
        print(res.text)
asyncio.run(run())
"""

httpx 传二进制body参数名是 content，不是data！

三、aiohttp（异步首选，高并发场景）

异步生态标准http客户端，爬虫、异步微服务大量使用

import aiohttp
import asyncio
import json

async def main():
    url = "http://127.0.0.1:8000/api/upload"
    payload = {"data": list(range(2000))}
    raw = json.dumps(payload).encode("utf-8")
    compressed_body = gzip_compress(raw)

    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(await resp.text())

asyncio.run(main())

四、pycurl（底层封装libcurl，高性能）

基于c库libcurl，吞吐性能极强，适合超高并发场景；缺点：windows安装麻烦、api晦涩

import pycurl
import json
from io import bytesio

url = "http://127.0.0.1:8000/api/upload"
payload = {"data": list(range(2000))}
raw = json.dumps(payload).encode("utf-8")
compressed_body = gzip_compress(raw)

resp_buffer = bytesio()
c = pycurl.curl()
c.setopt(c.url, url)
c.setopt(c.post, 1)
c.setopt(c.postfields, compressed_body)
c.setopt(c.writedata, resp_buffer)

headers = [
    "content-type: application/json; charset=utf-8",
    "content-encoding: gzip"
]
c.setopt(c.httpheader, headers)
c.perform()
c.close()

print(resp_buffer.getvalue().decode())

五、tornado.httpclient（tornado框架内置异步客户端）

如果你项目本身使用tornado web框架，可以直接使用内置客户端，无需额外依赖

import json
from tornado.httpclient import asynchttpclient, httprequest
import tornado.ioloop

async def fetch():
    url = "http://127.0.0.1:8000/api/upload"
    payload = {"data": list(range(2000))}
    raw = json.dumps(payload).encode("utf-8")
    compressed_body = gzip_compress(raw)

    req = httprequest(
        url,
        method="post",
        body=compressed_body,
        headers={
            "content-type": "application/json; charset=utf-8",
            "content-encoding": "gzip"
        }
    )
    client = asynchttpclient()
    res = await client.fetch(req)
    print(res.body.decode())

tornado.ioloop.ioloop.current().run_sync(fetch)