🐍 第15课：HTTP请求

—— requests：HTTP 的人类化接口

📌 本课目标

1️⃣ GET 请求

import requests

# 基本 GET
resp = requests.get("https://httpbin.org/get")
print(resp.status_code)   # 200
print(resp.headers["content-type"])
print(resp.json())        # 自动解析 JSON

# 带查询参数
params = {
    "name": "张三",
    "page": 1,
    "limit": 20,
}
resp = requests.get("https://httpbin.org/get", params=params)
# 实际请求: https://httpbin.org/get?name=张三&page=1&limit=20

# 自定义请求头
headers = {
    "User-Agent": "MyApp/1.0",
    "Accept": "application/json",
    "Authorization": "Bearer token123",
}
resp = requests.get("https://httpbin.org/get", headers=headers)

# 超时控制
resp = requests.get("https://httpbin.org/get", timeout=10)  # 10秒超时
resp = requests.get("https://httpbin.org/get", timeout=(3, 10))  # 连接3s，读取10s

2️⃣ POST 请求

import requests

# JSON 数据
data = {"name": "张三", "age": 28}
resp = requests.post("https://httpbin.org/post", json=data)
# 自动设置 Content-Type: application/json

# 表单数据
form_data = {"username": "admin", "password": "secret"}
resp = requests.post("https://httpbin.org/post", data=form_data)
# Content-Type: application/x-www-form-urlencoded

# 文件上传
with open("report.pdf", "rb") as f:
    files = {"file": ("report.pdf", f, "application/pdf")}
    resp = requests.post("https://httpbin.org/post", files=files)

# 多字段+多文件
files = [
    ("files", ("a.txt", open("a.txt", "rb"))),
    ("files", ("b.txt", open("b.txt", "rb"))),
]

3️⃣ 文件下载

import requests
from pathlib import Path

# 小文件下载
resp = requests.get("https://example.com/file.zip")
Path("file.zip").write_bytes(resp.content)

# 大文件流式下载（不占内存）
def download_file(url, save_path, chunk_size=8192):
    """流式下载大文件"""
    resp = requests.get(url, stream=True)
    resp.raise_for_status()

    total = int(resp.headers.get("content-length", 0))
    downloaded = 0

    with open(save_path, "wb") as f:
        for chunk in resp.iter_content(chunk_size=chunk_size):
            f.write(chunk)
            downloaded += len(chunk)
            if total:
                pct = downloaded / total * 100
                print(f"\r下载进度: {pct:.1f}%", end="")

    print(f"\n✅ 下载完成: {save_path}")
    return save_path

# 带进度条的下载
def download_with_progress(url, save_path):
    resp = requests.get(url, stream=True)
    total = int(resp.headers.get("content-length", 0))
    with open(save_path, "wb") as f:
        for chunk in resp.iter_content(chunk_size=8192):
            f.write(chunk)

4️⃣ Session 与 Cookie

import requests

# Session 保持会话
session = requests.Session()

# 设置全局请求头
session.headers.update({
    "User-Agent": "MyApp/1.0",
    "Authorization": "Bearer token123",
})

# 登录
login_resp = session.post("https://api.example.com/login",
    json={"username": "admin", "password": "secret"}
)

# 后续请求自动携带 Cookie
profile = session.get("https://api.example.com/profile")
data = session.get("https://api.example.com/data")

# 使用完关闭
session.close()

# 或使用 with
with requests.Session() as s:
    s.get("https://httpbin.org/cookies/set?name=value")
    resp = s.get("https://httpbin.org/cookies")
    print(resp.json())

5️⃣ 重试与错误处理

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

# 配置重试策略
retry_strategy = Retry(
    total=3,               # 最多重试3次
    backoff_factor=1,      # 退避因子: 1s, 2s, 4s
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST"],
)

adapter = HTTPAdapter(max_retries=retry_strategy)
session = requests.Session()
session.mount("https://", adapter)
session.mount("http://", adapter)

# 使用
resp = session.get("https://httpbin.org/get")

# 错误处理最佳实践
def safe_request(url, method="GET", **kwargs):
    """带完整错误处理的请求"""
    try:
        resp = requests.request(method, url, timeout=10, **kwargs)
        resp.raise_for_status()
        return resp
    except requests.ConnectionError:
        print("网络连接失败")
    except requests.Timeout:
        print("请求超时")
    except requests.HTTPError as e:
        print(f"HTTP错误: {e.response.status_code}")
    except requests.RequestException as e:
        print(f"请求异常: {e}")
    return None

6️⃣ 验证脚本

#!/usr/bin/env python3
"""第15课 HTTP请求验证"""
import requests

def test_get_request():
    """GET请求测试"""
    resp = requests.get("https://httpbin.org/get",
                        params={"key": "value"}, timeout=10)
    assert resp.status_code == 200
    data = resp.json()
    assert data["args"]["key"] == "value"
    print("✅ GET请求测试通过")

def test_post_json():
    """POST JSON测试"""
    resp = requests.post("https://httpbin.org/post",
                         json={"name": "张三"}, timeout=10)
    assert resp.status_code == 200
    data = resp.json()
    assert data["json"]["name"] == "张三"
    print("✅ POST JSON测试通过")

def test_post_form():
    """POST表单测试"""
    resp = requests.post("https://httpbin.org/post",
                         data={"user": "admin"}, timeout=10)
    assert resp.status_code == 200
    data = resp.json()
    assert data["form"]["user"] == "admin"
    print("✅ POST表单测试通过")

def test_headers():
    """自定义请求头测试"""
    resp = requests.get("https://httpbin.org/headers",
                        headers={"X-Custom": "test123"}, timeout=10)
    assert resp.status_code == 200
    data = resp.json()
    assert data["headers"]["X-Custom"] == "test123"
    print("✅ 自定义请求头测试通过")

def test_session():
    """Session测试"""
    with requests.Session() as s:
        s.get("https://httpbin.org/cookies/set?session_id=abc", timeout=10)
        resp = s.get("https://httpbin.org/cookies", timeout=10)
        assert resp.status_code == 200
    print("✅ Session测试通过")

if __name__ == "__main__":
    test_get_request()
    test_post_json()
    test_post_form()
    test_headers()
    test_session()
    print("\n🎉 第15课全部验证通过！")

🔑 本课要点