第四十四篇：常用内建模块（下）—— itertools、contextlib、urllib

前言

Python 的标准库中有很多被低估的宝藏模块。itertools 提供高效的迭代器工具，contextlib 简化上下文管理器编写，urllib 用于网络请求处理。本文将系统讲解这三大实用模块。

itertools —— 迭代器工具库

无限迭代器

import itertools

# count — 无限计数器
counter = itertools.count(start=10, step=5)
print(next(counter))  # 10
print(next(counter))  # 15
print(next(counter))  # 20
print(next(counter))  # 25
# 注意：不会停止！需配合 limit 使用

# cycle — 无限循环
colors = ['red', 'green', 'blue']
color_cycle = itertools.cycle(colors)
for _ in range(7):
    print(next(color_cycle), end=' ')
# red green blue red green blue red

# repeat — 重复某个值
for value in itertools.repeat('★', 5):
    print(value, end='')
# ★★★★★

有限迭代器组合

import itertools

# product — 笛卡尔积
pairs = list(itertools.product('AB', '12'))
print(pairs)
# [('A', '1'), ('A', '2'), ('B', '1'), ('B', '2')]

# 带重复选择的笛卡尔积
powers_of_two = list(itertools.product([0, 1], repeat=4))
print(powers_of_two[:8])
# [(0, 0, 0, 0), (0, 0, 0, 1), ..., (1, 1, 1, 1)]

# permutations — 全排列
perm = list(itertools.permutations('ABC', 2))
print(perm)
# [('A', 'B'), ('A', 'C'), ('B', 'A'), ('B', 'C'), ('C', 'A'), ('C', 'B')]

# combinations — 组合（不重复顺序）
comb = list(itertools.combinations('ABCD', 2))
print(comb)
# [('A', 'B'), ('A', 'C'), ('A', 'D'), ('B', 'C'), ('B', 'D'), ('C', 'D')]

# combinations_with_replacement — 可重复组合
comb_rep = list(itertools.combinations_with_replacement('ABC', 2))
print(comb_rep)
# [('A', 'A'), ('A', 'B'), ('A', 'C'), ('B', 'B'), ('B', 'C'), ('C', 'C')]

分组与过滤

import itertools

# chain — 扁平化嵌套序列
nested = [[1, 2, 3], [4, 5], [6, 7, 8]]
flattened = list(itertools.chain.from_iterable(nested))
print(flattened)  # [1, 2, 3, 4, 5, 6, 7, 8]

# groupby — 按条件分组（需要预先排序！）
from operator import itemgetter

people = sorted(
    [('Alice', 'F'), ('Bob', 'M'), ('Charlie', 'M'), ('Diana', 'F')],
    key=lambda x: x[1]
)
for gender, group in itertools.groupby(people, key=itemgetter(1)):
    members = list(group)
    print(f"{gender}: {[p[0] for p in members]}")
# F: ['Alice', 'Diana']
# M: ['Bob', 'Charlie']

# filterfalse — 保留不符合条件的元素
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
odds = list(itertools.filterfalse(lambda x: x % 2 == 0, numbers))
print(odds)  # [1, 3, 5, 7, 9]

# takewhile / dropwhile — 有条件地取元素
scores = [95, 88, 72, 65, 91, 55, 40]
passed = list(itertools.takewhile(lambda x: x >= 60, scores))
print(passed)  # [95, 88, 72, 65]

dropped = list(itertools.dropwhile(lambda x: x >= 60, scores))
print(dropped)  # [55, 40]

# islice — 切片迭代器（惰性）
def gen_primes():
    """生成素数"""
    yield 2
    candidate = 3
    while True:
        if all(candidate % p != 0 for p in range(2, int(candidate**0.5) + 1)):
            yield candidate
        candidate += 2

primes = list(itertools.islice(gen_primes(), 10))
print(primes)  # [2, 3, 5, 7, 11, 13, 17, 19, 23, 29]

实际应用：数据分块

import itertools

def chunk_data(data, chunk_size):
    """将数据分成固定大小的块"""
    iterator = iter(data)
    return list(iter(lambda: tuple(itertools.islice(iterator, chunk_size)), ()))

items = list(range(17))
chunks = chunk_data(items, 5)
print(chunks)
# [(0, 1, 2, 3, 4), (5, 6, 7, 8, 9), (10, 11, 12, 13, 14), (15, 16)]

# 实际用例：批量发送 API 请求（每次最多 100 条）
api_payloads = chunk_data(range(1000), 100)
for i, batch in enumerate(api_payloads):
    print(f"发送批次 {i+1}，包含 {len(batch)} 条记录")

contextlib —— 上下文管理器增强

@contextmanager 装饰器

import contextlib
import time

@contextlib.contextmanager
def timer(label="操作"):
    """计时上下文管理器"""
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"[{label}] 耗时: {elapsed:.4f}s")

with timer("计算平方和"):
    result = sum(i * i for i in range(1_000_000))

contextlib.suppress —— 忽略异常

import contextlib
import os

# 优雅地忽略特定异常
with contextlib.suppress(FileNotFoundError):
    os.remove("nonexistent_file.txt")

# 多类型异常
with contextlib.suppress(FileNotFoundError, PermissionError, OSError):
    os.remove("protected_file.txt")

contextlib.nullcontext —— 条件上下文管理

import contextlib

use_cache = False

if use_cache:
    cache_context = open("cache.json", "r")
else:
    cache_context = contextlib.nullcontext()

with cache_context as f:
    if f:
        pass  # 加载缓存数据
    else:
        pass  # 使用默认配置

ExitStack 动态添加上下文

import contextlib

# 动态打开多个文件
def process_files(file_paths):
    with contextlib.ExitStack() as stack:
        files = [stack.enter_context(open(fp)) for fp in file_paths]
        results = [f.read() for f in files]
        return results

empty_results = process_files([])
print(f"空文件列表结果: {empty_results}")

数据库事务管理器实战

import contextlib

class FakeDatabase:
    def __init__(self):
        self.data = {}
        self.rollback_log = []
    
    def begin_transaction(self):
        self._snapshot = self.data.copy()
        print("[DB] 开始事务")
    
    def commit(self):
        print("[DB] 提交事务 ✓")
    
    def rollback(self):
        self.data = self._snapshot.copy()
        print("[DB] 回滚事务")
    
    def execute(self, sql):
        if "INSERT" in sql:
            parts = sql.split("'")
            self.data[parts[1]] = parts[3]
        elif "DELETE" in sql:
            key = sql.split("'")[1]
            self.data.pop(key, None)

db = FakeDatabase()

@contextlib.contextmanager
def transaction(database):
    database.begin_transaction()
    try:
        yield database
    except Exception as e:
        print(f"[ERROR] {e}")
        database.rollback()
        raise
    else:
        database.commit()

# 成功的事务
with transaction(db) as conn:
    conn.execute("INSERT 'user' 'Alice'")
    conn.execute("INSERT 'email' 'alice@example.com'")

print(f"数据: {db.data}")  # {'user': 'Alice', 'email': 'alice@example.com'}

# 失败的事务
try:
    with transaction(db) as conn:
        conn.execute("INSERT 'temp' 'data'")
        raise ValueError("模拟数据库错误!")
except ValueError:
    print("事务已回滚!")

print(f"回滚后数据: {db.data}")  # 没有 temp 数据

urllib —— 网络请求模块

HTTP GET 请求

import urllib.request
import urllib.error
import urllib.parse
import json

# 基本 GET 请求
url = "https://httpbin.org/get"
try:
    response = urllib.request.urlopen(url, timeout=10)
    print(f"状态码: {response.status}")  # 200
    print(f"Content-Type: {response.headers['Content-Type']}")
    data = json.loads(response.read().decode('utf-8'))
    print(f"URL: {data['url']}")
except urllib.error.URLError as e:
    if hasattr(e, 'code'):
        print(f"HTTP 错误 {e.code}: {e.reason}")
    else:
        print(f"连接错误: {e.reason}")

# 带查询参数的 URL
params = urllib.parse.urlencode({"q": "python tutorial", "page": 1})
full_url = f"https://httpbin.org/get?{params}"

POST 请求

import urllib.request
import urllib.parse
import json

# POST JSON 数据
json_data = json.dumps({"username": "alice", "password": "secret123"}).encode('utf-8')
req = urllib.request.Request(
    "https://httpbin.org/post",
    data=json_data,
    headers={"Content-Type": "application/json"},
    method="POST"
)

try:
    response = urllib.request.urlopen(req, timeout=10)
    result = json.loads(response.read().decode('utf-8'))
    print(f"接收到的 JSON: {result['json']}")
except urllib.error.HTTPError as e:
    print(f"HTTP 错误 {e.code}: {e.reason}")

# POST 表单数据
form_data = urllib.parse.urlencode({"username": "alice", "password": "secret"}).encode('utf-8')
form_req = urllib.request.Request(
    "https://httpbin.org/post",
    data=form_data,
    headers={"Content-Type": "application/x-www-form-urlencoded"},
    method="POST"
)

urllib.parse URL 工具

import urllib.parse

# URL 解析
url = "https://www.example.com:8080/path/to/page?id=123&name=test#section"
parsed = urllib.parse.urlparse(url)

print(f"协议:     {parsed.scheme}")        # https
print(f"域名:     {parsed.netloc}")         # www.example.com:8080
print(f"路径:     {parsed.path}")           # /path/to/page
print(f"查询参数: {parsed.query}")          # id=123&name=test
print(f"片段:     {parsed.fragment}")       # section
print(f"端口:     {parsed.port}")           # 8080

# URL 编码/解码
search_query = "Python 教程 & 最佳实践"
encoded = urllib.parse.quote(search_query)
print(f"编码后: {encoded}")
decoded = urllib.parse.unquote(encoded)
print(f"解码后: {decoded}")

# 查询参数解析
query_string = "id=123&name=测试&page=2"
params = urllib.parse.parse_qs(query_string)
print(f"解析参数: {params}")

# 构建新 URL
new_params = {"id": "456", "name": "新名称"}
base_url = "https://example.com/search"
full_url = f"{base_url}?{urllib.parse.urlencode(new_params)}"
print(f"新 URL: {full_url}")

HTTP 状态码速查表

状态码	含义	说明
200	OK	请求成功
301	Moved Permanently	永久重定向
302	Found	临时重定向
400	Bad Request	客户端请求错误
401	Unauthorized	未授权
403	Forbidden	禁止访问
404	Not Found	页面不存在
500	Internal Server Error	服务器内部错误
502	Bad Gateway	网关错误
503	Service Unavailable	服务不可用

最佳实践总结

✅ 推荐做法

itertools 保持惰性 — 延迟求值，内存友好
contextlib.contextmanager — 比写类更简洁地创建上下文管理器
urllib 始终设置 timeout — 防止永久等待
URLError 处理区分网络层和 HTTP 层 — 更精确的错误处理
urllib.parse 而非手动拼接 URL — 正确处理特殊字符编码
requests 替代方案 — 对于复杂 HTTP 需求，考虑安装 requests 库

❌ 避免做法

import urllib.request
import urllib.error

# ❌ 不处理超时
response = urllib.request.urlopen("http://example.com/slow-endpoint")

# ✅ 始终设置 timeout
response = urllib.request.urlopen("http://example.com/slow-endpoint", timeout=10)

# ❌ 直接拼接 URL（不安全）
url = f"https://example.com/api?q={user_input}"

# ✅ 使用 urlencode
safe_url = f"https://example.com/api?{urllib.parse.urlencode({'q': user_input})}"

# ❌ 忽略 URLError
try:
    response = urllib.request.urlopen("http://broken-url")
except:
    pass  # 丢失错误信息!

# ✅ 明确处理异常
try:
    response = urllib.request.urlopen("http://broken-url", timeout=5)
except urllib.error.URLError as e:
    print(f"网络错误: {e.reason}")

总结

本文系统讲解了 Python 三大实用标准库模块：

itertools — count/cycle/repeat 无限迭代器，product/permutations/combinations 组合工具，chain/groupby/filterfalse/islice 过滤工具
contextlib — @contextmanager 创建上下文管理器，suppress/nullcontext/ExitStack 实用工具
urllib — urlopen 发起请求，Request 自定义头部，parse URL 编码/解码
实战应用 — 数据分块、事务管理、URL 工具
最佳实践 — 超时设置、URL 安全编码、异常分类处理

本文是 Python 教程系列的第四十四篇。