7 个 Python 库让你的网络编程脱胎换骨
请求一多,程序先死的往往不是业务逻辑,是网络这一层。
最常见的味道就是这几种:requests.get() 写满项目、超时不设、重试乱补、连接池没有、抓包时一脸懵。代码能跑,但线上一抖就全露馅。Python 做网络编程,不是会发 HTTP 请求就算完,真到接口风暴、批量采集、异步推送、代理切换、WebSocket 长连这些场景,库选不对,后面全是补丁。
下面这 7 个库,我自己看下来,属于那种“平时不用觉得没事,一旦用上就回不去”的类型。
1. requests:别嫌老,很多活它还真最顺手
这东西没什么可神化的,但也别动不动就说过时。内部系统对接、后台脚本、临时排障,requests 还是顺手。
import requestssession = 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 httpxasyncdeffetch_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 webasyncdefhealth(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 jsonasyncdefhandler(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 Retryretries = 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, sr1pkt = 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 解决的是“出事后别瞎猜”。
网络编程这活,库不是越新越好,也不是越多越专业。关键是你心里要有顺序:先把连接、超时、重试、并发模型搞对,再谈优雅。 不然代码看着挺现代,请求一超时,还是老毛病。