当今的软件世界是互联的。你的程序需要获取天气数据、调用 ChatGPT API、从 GitHub 拉取仓库信息、向飞书群发送通知……所有这些都离不开网络请求。

Python 的标准库提供了 urllib.request，但社区几乎都选择更人性化的 requests 库。本章从基础到实战，带你掌握 Python 中的 HTTP 通信。

HTTP 基础回顾

在你写代码之前，了解 HTTP 的基本概念会很有帮助：

概念 说明
URL 统一资源定位符，标识资源的位置
HTTP 方法 GET（获取）、POST（创建）、PUT（全量更新）、PATCH（部分更新）、DELETE（删除）
请求头 携带元信息，如认证 Token、内容类型
请求体 POST/PUT 时发送的数据
状态码 200（成功）、201（创建成功）、400（请求错误）、401（未认证）、404（未找到）、500（服务器错误）
响应体 服务器返回的数据，通常为 JSON 格式

urllib.request：内置方案

Python 标准库自带的 HTTP 客户端，无需安装：

from urllib.request import Request, urlopen
from urllib.parse import urlencode
import json

# GET 请求
response = urlopen("https://httpbin.org/get")
print(response.status)          # 200
print(response.read().decode()) # JSON 格式的响应内容

# POST 请求
data = urlencode({"name": "Alice", "age": 30}).encode()
req = Request("https://httpbin.org/post", data=data, method="POST")
req.add_header("Content-Type", "application/x-www-form-urlencoded")

with urlopen(req) as resp:
    result = json.loads(resp.read().decode())
    print(result["form"])  # {'name': 'Alice', 'age': '30'}

比较一下 urllib.request 和 requests 的代码量：

操作 urllib.request requests
GET 请求 3 行 1 行
POST JSON 7 行 1 行
错误处理 手动检查状态码 raise_for_status()
Session 手动 build_opener requests.Session
文件上传 构建 multipart files={"file": open("f.txt")}

虽然标准库能完成所有工作，但每次都要手动编码参数、解码 JSON、处理错误……用 requests 会更舒心。

requests 库入门

安装

pip install requests

requests 是 Python 生态中下载量最大的包之一，已经成为 HTTP 请求的"事实标准"。

基础 GET 请求

import requests

response = requests.get("https://api.github.com")
print(response.status_code)   # 200
print(response.headers)        # 响应头字典
print(response.text)           # 原始文本内容
print(response.json())         # JSON 解析后的数据（返回 dict/list）

# 检查请求是否成功
print(response.ok)             # True（status_code < 400）

带参数的 GET 请求

# 方式一：手动拼接 URL（不推荐）
response = requests.get("https://httpbin.org/get?q=python&page=1")

# 方式二：params 参数（推荐）
params = {
    "q": "python",
    "page": 1,
    "sort": "stars",
    "order": "desc",
}
response = requests.get("https://api.github.com/search/repositories", params=params)
print(response.url)
# https://api.github.com/search/repositories?q=python&page=1&sort=stars&order=desc

params 参数会自动进行 URL 编码，处理特殊字符：

params = {"q": "python testing", "lang": "zh-CN"}
response = requests.get("https://httpbin.org/get", params=params)
print(response.url)
# https://httpbin.org/get?q=python+testing&lang=zh-CN

POST 请求

import requests

# POST JSON 数据
data = {"title": "测试文章", "content": "这是内容"}
response = requests.post("https://httpbin.org/post", json=data)
print(response.status_code)   # 200
print(response.json()["json"])  # 回显发送的 JSON

# POST 表单数据
form_data = {"username": "alice", "password": "secret"}
response = requests.post("https://httpbin.org/post", data=form_data)
print(response.json()["form"])  # {'username': 'alice', 'password': 'secret'}

关键区别：json= 自动设置 Content-Type: application/json 并将数据序列化；data= 设置 Content-Type: application/x-www-form-urlencoded。

PUT / PATCH / DELETE

# PUT：全量更新
response = requests.put(
    "https://api.example.com/users/1",
    json={"name": "Alice", "email": "alice@example.com"}
)

# PATCH：部分更新
response = requests.patch(
    "https://api.example.com/users/1",
    json={"name": "Alice New Name"}
)

# DELETE：删除
response = requests.delete("https://api.example.com/users/1")

请求头与认证

自定义请求头

headers = {
    "User-Agent": "MyApp/1.0",
    "Accept": "application/json",
    "Authorization": "Bearer github_pat_xxxxx",
}
response = requests.get(
    "https://api.github.com/user",
    headers=headers
)

基本认证

from requests.auth import HTTPBasicAuth

response = requests.get(
    "https://api.github.com/user",
    auth=HTTPBasicAuth("username", "token_or_password")
)

# 简写方式
response = requests.get(
    "https://api.github.com/user",
    auth=("username", "token_or_password")
)

Bearer Token 认证

# 方式一：手动设置请求头
headers = {"Authorization": f"Bearer {token}"}

# 方式二：使用 auth 参数
class BearerAuth(requests.auth.AuthBase):
    """自定义认证类"""
    def __init__(self, token):
        self.token = token

    def __call__(self, r):
        r.headers["Authorization"] = f"Bearer {self.token}"
        return r

response = requests.get(
    "https://api.example.com/protected",
    auth=BearerAuth("your-token-here")
)

API Key 认证

# 方式一：请求头
headers = {"X-API-Key": "your-api-key"}

# 方式二：查询参数
params = {"api_key": "your-api-key", "city": "Beijing"}

response = requests.get("https://api.weather.example.com/current", params=params)

会话与会话池

为什么需要 Session？

每次 requests.get() 都会创建新的 TCP 连接，用完就关闭。如果有多个请求到同一个服务器，重复创建连接非常低效。Session 会复用 TCP 连接池：

import requests

# 不使用 Session——每次请求都新建连接
for _ in range(10):
    requests.get("https://api.github.com/")  # 10 次 TCP 握手

# 使用 Session——复用连接池
with requests.Session() as session:
    for _ in range(10):
        session.get("https://api.github.com/")  # 1 次 TCP 握手 + 9 次复用

Session 的其他好处

with requests.Session() as session:
    # 设置默认请求头——所有请求自动携带
    session.headers.update({
        "User-Agent": "MyApp/1.0",
        "Accept": "application/json",
    })

    # 设置默认基础 URL
    # 通过适配器实现
    session.mount("https://api.github.com", requests.adapters.HTTPAdapter())

    # 这些请求共享相同的请求头和连接池
    resp1 = session.get("https://api.github.com/user", auth=("user", "token"))
    resp2 = session.get("https://api.github.com/repos/psf/requests")
    resp3 = session.get("https://api.github.com/search/repositories", params={"q": "python"})

连接池设置

import requests
from requests.adapters import HTTPAdapter

session = requests.Session()

# 配置连接池：最多保持 10 个连接，每个主机最多 5 个
adapter = HTTPAdapter(
    pool_connections=10,
    pool_maxsize=10,
    pool_block=False,
)
session.mount("https://", adapter)
session.mount("http://", adapter)

超时与重试

超时设置

不设置超时的请求可能会永远挂起：

# 好：设置超时
try:
    response = requests.get(
        "https://api.github.com/",
        timeout=5  # 5 秒超时
    )
except requests.Timeout:
    print("请求超时了！")

超时可以用元组分别设置连接超时和读取超时：

# (连接超时, 读取超时)
response = requests.get(
    "https://api.github.com/",
    timeout=(3.05, 10)  # 连接 3.05 秒，读取 10 秒
)

重试机制

from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter

# 配置重试策略
retry_strategy = Retry(
    total=3,                # 最多重试 3 次
    backoff_factor=1,       # 退避因子：1, 2, 4 秒
    status_forcelist=[500, 502, 503, 504],  # 这些状态码触发重试
    allowed_methods=["GET", "POST"],         # 允许重试的 HTTP 方法
)

# 创建适配器
adapter = HTTPAdapter(max_retries=retry_strategy)

# 挂载到 Session
session = requests.Session()
session.mount("https://", adapter)
session.mount("http://", adapter)

# 使用——自动重试
response = session.get("https://api.github.com/")
print(f"请求成功，状态码: {response.status_code}")

错误处理

常见异常

import requests
from requests.exceptions import (
    RequestException,
    HTTPError,
    ConnectionError,
    Timeout,
    TooManyRedirects,
)

def safe_request(url: str, **kwargs) -> dict | None:
    """安全的 HTTP 请求封装"""
    try:
        response = requests.get(url, timeout=10, **kwargs)
        response.raise_for_status()  # 状态码 4xx/5xx 时抛出 HTTPError
        return response.json()
    except HTTPError as e:
        print(f"HTTP 错误: {e.response.status_code} - {e}")
    except ConnectionError as e:
        print(f"连接错误: {e}")
    except Timeout as e:
        print(f"超时: {e}")
    except TooManyRedirects as e:
        print(f"重定向过多: {e}")
    except RequestException as e:
        print(f"其他请求错误: {e}")
    return None

# 使用
data = safe_request("https://api.github.com/users/octocat")
if data:
    print(f"用户: {data['login']}")

响应的条件检查

response = requests.get("https://api.github.com/")

# 更精细的状态码处理
if response.status_code == 200:
    data = response.json()
elif response.status_code == 404:
    print("资源不存在")
elif response.status_code == 403:
    print("被限流或被禁止访问")
elif response.status_code == 500:
    print("服务器内部错误")
else:
    print(f"未预期的状态码: {response.status_code}")

实战：调用 GitHub API

GitHub 的 REST API 是学习 HTTP 请求的绝佳资源——它公开、文档完善、不需要复杂认证就能使用。

获取用户信息

import requests

def get_github_user(username: str) -> dict:
    """获取 GitHub 用户信息"""
    response = requests.get(
        f"https://api.github.com/users/{username}",
        headers={"Accept": "application/vnd.github.v3+json"},
        timeout=10,
    )
    response.raise_for_status()
    return response.json()

# 使用
user = get_github_user("octocat")
print(f"用户名: {user['login']}")
print(f"仓库数: {user['public_repos']}")
print(f"粉丝数: {user['followers']}")
print(f"关注数: {user['following']}")
print(f"创建于: {user['created_at']}")

搜索仓库

import requests

def search_repos(query: str, per_page: int = 5) -> list[dict]:
    """搜索 GitHub 仓库"""
    response = requests.get(
        "https://api.github.com/search/repositories",
        params={
            "q": query,
            "sort": "stars",
            "order": "desc",
            "per_page": per_page,
        },
        headers={"Accept": "application/vnd.github.v3+json"},
        timeout=10,
    )
    response.raise_for_status()
    return response.json()["items"]

repos = search_repos("python web framework")
for repo in repos:
    print(f"⭐ {repo['full_name']} — {repo['stargazers_count']} stars")
    print(f"   {repo['description']}")
    print(f"   {repo['html_url']}")
    print()

需要认证的 API

import requests
import os

# 从环境变量获取 Token（不要 hardcode！）
token = os.environ.get("GITHUB_TOKEN")
if not token:
    raise RuntimeError("请设置 GITHUB_TOKEN 环境变量")

headers = {
    "Authorization": f"token {token}",
    "Accept": "application/vnd.github.v3+json",
}

session = requests.Session()
session.headers.update(headers)

# 获取认证用户信息
user = session.get("https://api.github.com/user", timeout=10).json()
print(f"认证用户: {user['login']}")

# 创建 Issue
issue_data = {
    "title": "发现一个 Bug",
    "body": "我在使用过程中发现了以下问题……",
    "labels": ["bug"],
}
response = session.post(
    "https://api.github.com/repos/octocat/Hello-World/issues",
    json=issue_data,
)
if response.status_code == 201:
    print(f"Issue 创建成功: {response.json()['html_url']}")

实战：获取天气数据

让我们用公开的 Open-Meteo API（无需 API Key）获取天气信息：

import requests
from datetime import datetime

def get_weather(city: str, lat: float, lon: float) -> dict:
    """从 Open-Meteo API 获取天气数据（无需 API Key）"""
    params = {
        "latitude": lat,
        "longitude": lon,
        "current": ["temperature_2m", "relative_humidity_2m", "weather_code"],
        "daily": ["temperature_2m_max", "temperature_2m_min"],
        "timezone": "auto",
        "forecast_days": 3,
    }

    response = requests.get(
        "https://api.open-meteo.com/v1/forecast",
        params=params,
        timeout=10,
    )
    response.raise_for_status()
    return response.json()

def display_weather(city: str, data: dict):
    """格式化显示天气信息"""
    current = data["current"]
    daily = data["daily"]

    print(f"🌍 {city} 天气")
    print(f"  当前温度: {current['temperature_2m']}°C")
    print(f"  湿度: {current['relative_humidity_2m']}%")
    print(f"  天气代码: {current['weather_code']}")
    print()
    print("  未来三天预报:")
    for i in range(len(daily["time"])):
        print(f"    {daily['time'][i]}: "
              f"{daily['temperature_2m_min'][i]}~{daily['temperature_2m_max'][i]}°C")

# 使用
weather = get_weather("北京", 39.90, 116.41)
display_weather("北京", weather)

实战：构建 API 客户端类

将远程 API 封装成类，代码更整洁、可复用：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from dataclasses import dataclass


@dataclass
class Repository:
    """GitHub 仓库的数据模型"""
    name: str
    full_name: str
    description: str | None
    stars: int
    forks: int
    language: str | None
    url: str


class GitHubClient:
    """GitHub API 客户端"""

    BASE_URL = "https://api.github.com"

    def __init__(self, token: str | None = None):
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/vnd.github.v3+json",
            "User-Agent": "PythonCourse/1.0",
        })

        if token:
            self.session.headers["Authorization"] = f"token {token}"

        # 配置重试
        retries = Retry(
            total=3,
            backoff_factor=0.5,
            status_forcelist=[500, 502, 503],
        )
        self.session.mount("https://", HTTPAdapter(max_retries=retries))

    def get_user(self, username: str) -> dict:
        """获取用户信息"""
        resp = self.session.get(f"{self.BASE_URL}/users/{username}", timeout=10)
        resp.raise_for_status()
        return resp.json()

    def search_repos(self, query: str, limit: int = 10) -> list[Repository]:
        """搜索仓库"""
        resp = self.session.get(
            f"{self.BASE_URL}/search/repositories",
            params={"q": query, "per_page": limit, "sort": "stars"},
            timeout=10,
        )
        resp.raise_for_status()
        items = resp.json()["items"]

        return [
            Repository(
                name=item["name"],
                full_name=item["full_name"],
                description=item.get("description"),
                stars=item["stargazers_count"],
                forks=item["forks_count"],
                language=item.get("language"),
                url=item["html_url"],
            )
            for item in items
        ]

    def get_repo_issues(self, owner: str, repo: str, state: str = "open") -> list[dict]:
        """获取仓库的 Issues"""
        resp = self.session.get(
            f"{self.BASE_URL}/repos/{owner}/{repo}/issues",
            params={"state": state, "per_page": 10},
            timeout=10,
        )
        resp.raise_for_status()
        return resp.json()

    def close(self):
        """关闭 Session"""
        self.session.close()


# 使用示例
def main():
    client = GitHubClient()

    try:
        # 搜索 Python 仓库
        repos = client.search_repos("python async framework", limit=5)
        print("热门 Python 异步框架：")
        for repo in repos:
            print(f"  ⭐ {repo.full_name} ({repo.stars} stars)")
            if repo.description:
                print(f"     {repo.description}")

    finally:
        client.close()


if __name__ == "__main__":
    main()

流式请求与下载大文件

流式下载

import requests

def download_large_file(url: str, local_path: str, chunk_size: int = 8192):
    """下载大文件（避免内存溢出）"""
    with requests.get(url, stream=True, timeout=30) as response:
        response.raise_for_status()
        with open(local_path, "wb") as f:
            for chunk in response.iter_content(chunk_size=chunk_size):
                if chunk:  # 过滤 keep-alive 的空 chunk
                    f.write(chunk)

    print(f"文件已保存到: {local_path}")

# 使用流式下载，即使文件很大也不会撑爆内存
download_large_file(
    "https://github.com/psf/requests/archive/refs/heads/main.zip",
    "requests-main.zip"
)

进度显示

import requests
from tqdm import tqdm  # pip install tqdm

def download_with_progress(url: str, local_path: str):
    """带进度条的下载"""
    response = requests.get(url, stream=True, timeout=30)
    response.raise_for_status()

    total_size = int(response.headers.get("content-length", 0))

    with (
        open(local_path, "wb") as f,
        tqdm(
            desc="下载中",
            total=total_size,
            unit="B",
            unit_scale=True,
        ) as progress,
    ):
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:
                f.write(chunk)
                progress.update(len(chunk))

速率限制（Rate Limiting）

大多数 API 会限制请求频率，超过限制会被临时封禁。

import requests
import time

class RateLimitedClient:
    """带速率限制的 API 客户端"""

    def __init__(self, requests_per_minute: int = 30):
        self.min_interval = 60.0 / requests_per_minute
        self.last_request_time = 0.0

    def _wait_if_needed(self):
        """确保请求间隔不低于最小间隔"""
        elapsed = time.time() - self.last_request_time
        if elapsed < self.min_interval:
            wait_time = self.min_interval - elapsed
            time.sleep(wait_time)

    def get(self, url: str, **kwargs) -> requests.Response:
        self._wait_if_needed()
        response = requests.get(url, **kwargs)
        self.last_request_time = time.time()

        # 检查剩余配额
        remaining = int(response.headers.get("X-RateLimit-Remaining", 1))
        if remaining == 0:
            reset_time = int(response.headers.get("X-RateLimit-Reset", 0))
            wait = max(0, reset_time - time.time())
            print(f"配额用尽，等待 {wait:.0f} 秒后重置")
            if wait > 0:
                time.sleep(wait)

        return response


# GitHub 未认证用户每小时 60 次请求
client = RateLimitedClient(requests_per_minute=30)

# 安全地批量请求
for username in ["octocat", "torvalds", "defunkt"]:
    resp = client.get(f"https://api.github.com/users/{username}")
    data = resp.json()
    print(f"{data['login']}: {data['public_repos']} 个仓库")

使用 httpx：异步 HTTP 客户端

如果你想在异步代码中发送 HTTP 请求，httpx 是更好的选择：

pip install httpx

import httpx
import asyncio

async def fetch_github_user(username: str) -> dict:
    """异步获取 GitHub 用户信息"""
    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.get(
            f"https://api.github.com/users/{username}",
            headers={"Accept": "application/vnd.github.v3+json"},
        )
        response.raise_for_status()
        return response.json()

async def main():
    """并发获取多个用户信息"""
    users = await asyncio.gather(
        fetch_github_user("octocat"),
        fetch_github_user("torvalds"),
        fetch_github_user("defunkt"),
    )

    for user in users:
        print(f"{user['login']}: {user['public_repos']} 个仓库")

# 运行
asyncio.run(main())

httpx 的 API 设计和 requests 几乎一样，但支持异步。如果项目已经用 asyncio，优先选择 httpx。

网络请求最佳实践

1. 总是设置超时

# 坏：可能永远挂起
requests.get("https://api.example.com/")

# 好：有超时保护
requests.get("https://api.example.com/", timeout=10)

2. 使用 Session 复用连接

# 坏：每次创建新连接
for url in urls:
    requests.get(url)

# 好：复用连接池
with requests.Session() as session:
    for url in urls:
        session.get(url)

3. 对 4xx/5xx 状态码显式处理

# 要么 raise 异常
response.raise_for_status()

# 要么手动检查
if response.status_code != 200:
    handle_error(response)

4. 不要硬编码敏感信息

# 坏：Token 写在代码里
token = "ghp_xxxxxxxxxxxx"

# 好：从环境变量读取
import os
token = os.environ.get("API_TOKEN")
if not token:
    raise RuntimeError("请设置 API_TOKEN 环境变量")

5. 遵守 API 的速率限制

# 检查响应头
remaining = response.headers.get("X-RateLimit-Remaining", "unknown")
print(f"剩余配额: {remaining}")

# 如果被限制，等待后再重试
if response.status_code == 429:
    retry_after = int(response.headers.get("Retry-After", 60))
    time.sleep(retry_after)

6. 记录请求日志

import logging

# 开启 requests 的调试日志
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("urllib3").setLevel(logging.DEBUG)

# 或者使用 requests 的钩子记录
def log_request(response, *args, **kwargs):
    print(f"[{response.status_code}] {response.request.method} {response.request.url}")

session = requests.Session()
session.hooks["response"].append(log_request)

小结

本章我们学习了 Python 网络请求的核心知识：

  • HTTP 基础：GET/POST/PUT/DELETE、请求头、状态码、JSON 响应
  • urllib.request：标准库内置方案，但语法繁琐
  • requests 库：事实标准的 HTTP 客户端，get()/post() 一行搞定
  • Session：复用 TCP 连接池，设置默认头，性能提升明显
  • 超时与重试：timeout 参数防挂起，urllib3.Retry 自动重试失败请求
  • 错误处理：raise_for_status() 抛出 HTTPError，try/except 捕获网络异常
  • 实战：调用 GitHub API 查询用户和仓库、获取天气数据、构建 API 客户端类
  • 速率限制：尊重 API 配额，超出后等待再请求
  • httpx：异步 HTTP 客户端，与 asyncio 配合使用

下一步： 最后一章，我们将综合运用所有知识，从零构建一个完整的命令行工具！

Summary: requests 库、HTTP 方法、Session、超时重试、API 调用实战。