Python技术迷

7 个 Python 库让你的网络编程脱胎换骨

请求一多,程序先死的往往不是业务逻辑,是网络这一层。

最常见的味道就是这几种:requests.get() 写满项目、超时不设、重试乱补、连接池没有、抓包时一脸懵。代码能跑,但线上一抖就全露馅。Python 做网络编程,不是会发 HTTP 请求就算完,真到接口风暴、批量采集、异步推送、代理切换、WebSocket 长连这些场景,库选不对,后面全是补丁。

下面这 7 个库,我自己看下来,属于那种“平时不用觉得没事,一旦用上就回不去”的类型。

1. requests:别嫌老,很多活它还真最顺手

这东西没什么可神化的,但也别动不动就说过时。内部系统对接、后台脚本、临时排障,requests 还是顺手。

import requests

session = requests.Session()
session.headers.update({"User-Agent": "ops-script/1.0"})

resp = session.get(
"https://api.internal.example.com/orders",
    params={"status": "paid"},
    timeout=(2, 5)   # 连接超时、读取超时
)

print(resp.status_code)
print(resp.json())

我一般第一眼就看两件事:有没有 Session,有没有 timeout。这两个没写,后面出慢请求,锅基本跑不掉。

2. httpx:想同步也行,想异步也行

requests 好用,但一到异步并发就不够看了。httpx 比较舒服的地方,是接口风格接近 requests,迁移成本低,顺手就能切异步。

import asyncio
import httpx

asyncdeffetch_user(client, user_id):
    r = await client.get(f"https://api.example.com/users/{user_id}")
return r.json()

asyncdefmain():
asyncwith httpx.AsyncClient(timeout=5.0) as client:
        data = await asyncio.gather(
            fetch_user(client, 101),
            fetch_user(client, 102),
            fetch_user(client, 103),
        )
        print(data)

asyncio.run(main())

批量查接口、聚合下游数据,这类代码用它比线程池干净得多。尤其是你已经在 asyncio 体系里,再硬塞 requests,看着就别扭。

3. aiohttp:异步 HTTP 老将,客户端服务端两头通吃

有些项目不是“发请求”这么简单,它本身就是个高并发网络服务。aiohttp 这种时候就比普通请求库更像工具。

from aiohttp import web

asyncdefhealth(request):
return web.json_response({"ok": True})

asyncdefpush_task(request):
    body = await request.json()
    print("收到任务:", body.get("task_id"))
return web.json_response({"accepted": True})

app = web.Application()
app.router.add_get("/health", health)
app.router.add_post("/push", push_task)

web.run_app(app, host="0.0.0.0", port=8080)

做轻量网关、回调接收服务、内部异步接口,这库很顶。不是说 Flask 不行,是并发模型和场景不一样。

4. websockets:长连接别自己硬撸协议

很多人第一次做消息推送,先拿 HTTP 轮询顶着,顶到后面延迟高、连接多、日志炸。真需要长连接,就老老实实上 WebSocket。

import asyncio
import websockets
import json

asyncdefhandler(ws):
asyncfor message in ws:
        data = json.loads(message)
await ws.send(json.dumps({
"event": "ack",
"trace_id": data.get("trace_id")
        }))

asyncdefmain():
asyncwith websockets.serve(handler, "0.0.0.0", 8765):
await asyncio.Future()

asyncio.run(main())

聊天室、设备状态推送、控制台实时日志,这种场景它很省事。自己拿 socket 拼协议,不是不能干,是后面排障会很烦。

5. urllib3:连接池、重试、底层控制更细

很多人用 requests,却不知道真正干活的是 urllib3。你要更细地控连接池和重试策略,它就该直接上场了。

import urllib3
from urllib3.util.retry import Retry

retries = Retry(
    total=3,
    backoff_factor=0.5,
    status_forcelist=[500, 502, 503, 504]
)

http = urllib3.PoolManager(
    num_pools=20,
    maxsize=50,
    retries=retries,
    timeout=urllib3.Timeout(connect=2.0, read=5.0)
)

resp = http.request("GET", "https://api.example.com/ping")
print(resp.status, resp.data.decode())

这里有个误区我顺手提一句:不是所有失败都该重试。POST 幂等没处理好,重试一次,账可能就扣两次。

6. tenacity:网络代码里,重试别手搓

线上最怕看到这种代码:

# 别这么写
for _ in range(3):
try:
        call_api()
break
except Exception:
pass

看着像重试,实际上日志没了、异常吞了、间隔没有,故障一来根本没法复盘。tenacity 这种库就是专门收拾这种烂尾代码的。

from tenacity import retry, stop_after_attempt, wait_exponential
import requests

@retry(stop=stop_after_attempt(4), wait=wait_exponential(multiplier=1, min=1, max=8))
defquery_inventory(sku):
    r = requests.get(
f"https://stock.example.com/items/{sku}",
        timeout=3
    )
    r.raise_for_status()
return r.json()

print(query_inventory("SKU-9001"))

网络抖动、临时 502、下游偶发超时,这种场景很适合。但还是那句话,重试之前先判断这个请求值不值得重试。

7. scapy:抓包、造包、验协议,排障的时候是真能救命

普通 CRUD 程序员平时未必碰它,但只要你做设备通信、内网协议排查、奇怪 TCP 问题,这玩意就不是“高级”,是“能不能继续查下去”。

from scapy.all import IP, TCP, sr1

pkt = IP(dst="192.168.1.10") / TCP(dport=8080, flags="S")
resp = sr1(pkt, timeout=2, verbose=0)

if resp:
    print("端口可达,返回标志位:", resp.sprintf("%TCP.flags%"))
else:
    print("没回应,先别急着怀疑业务代码")

有些网络问题,日志里只有一个超时,业务层什么都看不出来。这时候你再空谈“可能是网络原因”,没用,先抓证据。

这 7 个库里,requests 和 httpx 解决的是“怎么发得舒服”,aiohttp 和 websockets 解决的是“怎么撑住并发和长连接”,urllib3 和 tenacity 解决的是“怎么把不稳定收拾干净”,scapy 解决的是“出事后别瞎猜”。

网络编程这活,库不是越新越好,也不是越多越专业。关键是你心里要有顺序:先把连接、超时、重试、并发模型搞对,再谈优雅。 不然代码看着挺现代,请求一超时,还是老毛病。