Python3 网络编程
The socket API, TCP echo server, and calling HTTP APIs with requests/httpx.
Key PointsSockets are byte streams — you must frame messages yourself; always set timeouts on every network call.
The socket API
Python
import socket
# TCP 客户端
with socket.create_connection(("example.com", 80), timeout=5) as s:
s.sendall(b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n")
chunks = []
while chunk := s.recv(4096):
chunks.append(chunk)
print(b"".join(chunks)[:120])Python
# TCP 服务端：接受连接，回显数据
import socket
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as srv:
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(("0.0.0.0", 9000))
srv.listen(5)
conn, addr = srv.accept()
with conn:
while data := conn.recv(1024):
conn.sendall(data)TCP is a stream: a single recv() may return half a message or two messages. Always frame by length or delimiter before parsing.
HTTP with requests
Python
import requests
resp = requests.get(
"https://api.example.com/users",
params={"page": 1},
headers={"Authorization": "Bearer token"},
timeout=(3, 10), # (连接超时, 读超时)
)
resp.raise_for_status() # 4xx/5xx 抛异常，别忘
data = resp.json()
requests.post("https://api.example.com/users", json={"name": "alice"}, timeout=5)Session, Retry, Async
Python
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
session.mount("https://", HTTPAdapter(max_retries=Retry(
total=3, backoff_factor=0.5, status_forcelist=[502, 503, 504])))
# 异步版本（高并发场景）
import httpx
async def fetch(url):
async with httpx.AsyncClient(timeout=10) as client:
r = await client.get(url)
return r.json()Common Status Codes
| Code | Meaning | Action |
|---|---|---|
| 200 / 201 | OK / Created | Parse body |
| 301 / 302 | Redirect | Follow or update the URL |
| 400 | Bad request | Fix the payload, do not retry |
| 401 / 403 | Unauthenticated / forbidden | Refresh token or fix scopes |
| 429 | Rate limited | Back off using Retry-After |
| 500 / 502 / 503 | Server side failure | Retry with backoff |
socket API
Python
import socket
# TCP 客户端
with socket.create_connection(("example.com", 80), timeout=5) as s:
s.sendall(b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n")
chunks = []
while chunk := s.recv(4096):
chunks.append(chunk)
print(b"".join(chunks)[:120])Python
# TCP 服务端：接受连接，回显数据
import socket
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as srv:
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(("0.0.0.0", 9000))
srv.listen(5)
conn, addr = srv.accept()
with conn:
while data := conn.recv(1024):
conn.sendall(data)TCP 是字节流：一次 recv() 可能只拿到半条消息，也可能拿到两条。解析前必须按长度或分隔符分帧。
用 requests 发 HTTP
Python
import requests
resp = requests.get(
"https://api.example.com/users",
params={"page": 1},
headers={"Authorization": "Bearer token"},
timeout=(3, 10), # (连接超时, 读超时)
)
resp.raise_for_status() # 4xx/5xx 抛异常，别忘
data = resp.json()
requests.post("https://api.example.com/users", json={"name": "alice"}, timeout=5)会话、重试与异步
Python
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
session.mount("https://", HTTPAdapter(max_retries=Retry(
total=3, backoff_factor=0.5, status_forcelist=[502, 503, 504])))
# 异步版本（高并发场景）
import httpx
async def fetch(url):
async with httpx.AsyncClient(timeout=10) as client:
r = await client.get(url)
return r.json()常见状态码
| 状态码 | 含义 | 应对 |
|---|---|---|
| 200 / 201 | 成功 / 已创建 | 解析响应体 |
| 301 / 302 | 重定向 | 跟随或更新地址 |
| 400 | 请求错误 | 修参数，不要重试 |
| 401 / 403 | 未认证 / 无权限 | 刷新令牌或补权限 |
| 429 | 限流 | 按 Retry-After 退避 |
| 500 / 502 / 503 | 服务端故障 | 退避重试 |