Python►并发编程

HTTP 请求：requests 库入门

Python 内置有 urllib.request 能发 HTTP 请求，但 API 又老又难用。requests 库是 Python 生态里公认最好的 HTTP 客户端——“HTTP for Humans”，让发请求变得跟说话一样自然。这一篇讲清 requests 的日常使用：GET / POST、参数、headers、认证、超时、Session、以及处理响应的正确姿势。

一、安装

1
pip install requests

二、最简单的 GET

1
2
3
4
5
6
7
import requests

r = requests.get("https://httpbin.org/get")

print(r.status_code)     # 200
print(r.text)            # 响应内容（字符串）
print(r.json())          # 直接解析成 dict

三、常用响应属性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
r = requests.get(url)

r.status_code       # 200
r.reason            # 'OK'
r.headers           # 响应头字典
r.text              # 文本内容（str）
r.content           # 二进制内容（bytes）
r.json()            # JSON 解析
r.url               # 最终 URL（重定向后）
r.encoding          # 猜测的编码
r.cookies           # 服务器 set 的 cookies
r.elapsed           # 耗时（timedelta）
r.ok                # True 如果 status_code < 400
r.raise_for_status()   # 4xx/5xx 抛异常

四、URL 参数（GET）

1
2
3
4
5
6
7
8
9
10
# ❌ 手动拼
requests.get(f"https://api.com/search?q={q}&page=1")

# ✅ 用 params
r = requests.get("https://api.com/search", params={
    "q": "python",
    "page": 1,
    "tags": ["web", "async"],   # 会展开成多个 tags=web&tags=async
})
print(r.url)

requests 会自动 URL 编码。

五、POST 请求

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 表单
r = requests.post("https://httpbin.org/post", data={"name": "咖飞", "age": 3})

# JSON body（更常见）
r = requests.post("https://httpbin.org/post", json={"name": "咖飞", "age": 3})

# 上传文件
with open("photo.jpg", "rb") as f:
    r = requests.post("https://httpbin.org/post", files={"photo": f})

# 组合
r = requests.post(url,
    files={"photo": open("photo.jpg", "rb")},
    data={"caption": "咖飞的照片"},
)

data 和 json 的区别：

  • data=dict：application/x-www-form-urlencoded；
  • json=dict：application/json，自动 JSON 序列化，自动加 header。

六、headers

1
2
3
4
5
r = requests.get(url, headers={
    "User-Agent": "MyApp/1.0",
    "Accept": "application/json",
    "Authorization": "Bearer xxx",
})

七、超时

永远设置超时：

1
2
r = requests.get(url, timeout=5)         # 总超时 5 秒
r = requests.get(url, timeout=(3, 10))   # (连接超时, 读取超时)

不设的话可能被服务器”吊死”到永远。

八、错误处理

1
2
3
4
5
6
7
8
9
10
11
try:
    r = requests.get(url, timeout=5)
    r.raise_for_status()             # 4xx / 5xx 抛 HTTPError
except requests.Timeout:
    print("超时")
except requests.ConnectionError:
    print("连接失败")
except requests.HTTPError as e:
    print(f"HTTP {e.response.status_code}")
except requests.RequestException as e:
    print(f"其他 requests 异常：{e}")

requests.RequestException 是所有 requests 异常的根。

九、认证

1
2
3
4
5
6
7
# Basic Auth
r = requests.get(url, auth=("user", "password"))

# Bearer token
r = requests.get(url, headers={"Authorization": "Bearer xxx"})

# 复杂 auth 用 HTTPDigestAuth / HTTPProxyAuth 等

十、Session：复用连接和 cookies

1
2
3
4
5
6
7
8
9
with requests.Session() as s:
    # 登录：cookie 自动保存
    s.post("https://example.com/login", data={"user": "x", "pass": "y"})

    # 后续请求自动带 cookie
    r = s.get("https://example.com/profile")

    # 全局默认 headers
    s.headers.update({"User-Agent": "MyBot/1.0"})

好处：

  • 连接池复用（TCP 连接不用重新握手）；
  • 自动管理 cookies；
  • 统一 headers / auth。

批量请求同一个站点时永远用 Session——速度快 5-10 倍。

十一、重定向与代理

1
2
3
4
5
6
7
8
9
10
11
# 不跟随重定向
r = requests.get(url, allow_redirects=False)

# 代理
r = requests.get(url, proxies={
    "http": "http://127.0.0.1:7890",
    "https": "http://127.0.0.1:7890",
})

# SSL 校验（默认开，某些自签证书场景可关，但要谨慎）
r = requests.get(url, verify=False)

十二、流式下载大文件

1
2
3
4
5
with requests.get(url, stream=True) as r:
    r.raise_for_status()
    with open("bigfile.zip", "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)

stream=True 让 requests 不立刻下载整个 body，而是按块。

十三、上传大文件（流式）

1
2
3
# 内存不够时不能 `open(path).read()`
with open("bigfile.zip", "rb") as f:
    r = requests.post(url, data=f)   # 直接传文件对象

十四、超时重试

requests 本身没重试。用 urllib3.Retry 或自己写：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


def make_session():
    s = requests.Session()
    retry = Retry(
        total=3,
        backoff_factor=1,         # 1s, 2s, 4s
        status_forcelist=[500, 502, 503, 504],
        allowed_methods=["GET", "POST"],
    )
    adapter = HTTPAdapter(max_retries=retry)
    s.mount("http://", adapter)
    s.mount("https://", adapter)
    return s

或者用第三方 tenacity：

1
2
3
4
5
6
7
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
def fetch(url):
    r = requests.get(url, timeout=5)
    r.raise_for_status()
    return r

十五、异步 HTTP：httpx / aiohttp

requests 是同步库——一次请求会阻塞。想异步或高并发，用：

  • httpx：API 与 requests 兼容，支持同步和异步；
  • aiohttp：专注异步，更成熟。
1
2
3
4
5
6
7
# httpx 同步
import httpx
r = httpx.get(url)

# httpx 异步
async with httpx.AsyncClient() as client:
    r = await client.get(url)

具体见 异步编程 asyncio 入门。

十六、爬虫伦理与合规

用 requests 爬网站前请：

  • 读 robots.txt（有些站点禁止爬）；
  • 限速（time.sleep(1) 或用 asyncio.Semaphore）；
  • 加合理的 User-Agent（表明身份，不要伪装成浏览器）；
  • 不爬对方付费/需要认证的内容；
  • 频率过高会被封 IP，甚至法律风险。

十七、一个完整例子：GitHub API 爬取

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
# github.py
"""用 requests 从 GitHub API 获取用户仓库信息。"""

import requests
from pathlib import Path
import json
import time


class GitHubClient:
    BASE = "https://api.github.com"

    def __init__(self, token: str | None = None):
        self.s = requests.Session()
        self.s.headers.update({
            "Accept": "application/vnd.github+json",
            "User-Agent": "MyPythonScript/1.0",
        })
        if token:
            self.s.headers["Authorization"] = f"Bearer {token}"

    def get(self, path: str, params: dict = None) -> dict | list:
        r = self.s.get(f"{self.BASE}{path}", params=params, timeout=10)
        # 处理限流
        if r.status_code == 403 and "rate limit" in r.text.lower():
            reset = int(r.headers.get("X-RateLimit-Reset", 0))
            wait = max(0, reset - int(time.time()))
            print(f"限流，等待 {wait} 秒")
            time.sleep(wait + 1)
            return self.get(path, params)
        r.raise_for_status()
        return r.json()

    def user_repos(self, username: str) -> list:
        return self.get(f"/users/{username}/repos", params={"per_page": 100})

    def repo_readme(self, owner: str, repo: str) -> str:
        r = self.get(f"/repos/{owner}/{repo}/readme")
        # readme 内容是 base64
        import base64
        return base64.b64decode(r["content"]).decode("utf-8")


if __name__ == "__main__":
    import os
    client = GitHubClient(token=os.getenv("GITHUB_TOKEN"))
    repos = client.user_repos("python")
    for r in repos[:5]:
        print(f"{r['name']:30} ⭐ {r['stargazers_count']:>5} - {r.get('description', '')}")

十八、常见陷阱

  1. 忘设 timeout：请求永远卡住。
  2. 忘 raise_for_status：4xx/5xx 也会返回，容易误当成功。
  3. 循环里 requests.get 不用 Session：每次都握手，慢十倍。
  4. 下载大文件不用 stream：内存爆炸。
  5. response.text 编码不对：用 r.content.decode("utf-8") 显式解码，或先 r.encoding = "utf-8"。
  6. SSL 校验关了走公网：中间人攻击风险。除非确定内网自签，别关。
  7. json 参数值含 datetime：TypeError。要么先转字符串，要么自定义 json 序列化。

十九、小结与延伸阅读

  • 装 requests，用 get / post / put / delete；
  • URL 参数用 params；body 用 json 或 data；
  • 永远设 timeout；
  • 用 raise_for_status() 检查 HTTP 状态；
  • 批量请求用 Session（连接池 + cookies）；
  • 大文件用 stream=True + iter_content；
  • 重试用 Retry 或 tenacity；
  • 异步用 httpx 或 aiohttp。

延伸阅读：

下一篇 多线程 threading 基础 我们讲怎么并发做 IO 密集型任务。