Requests - HTTP 必备库

Python 最流行的 HTTP 客户端库。

安装

pip install requests

GET 请求

import requests

# 基本请求
r = requests.get("https://api.github.com")
print(r.status_code, r.json())

# URL 参数
r = requests.get("https://api.github.com/search/repos",
                  params={"q": "python", "sort": "stars"})

# 请求头
r = requests.get("https://api.github.com/user",
                 headers={"Authorization": "token xxx"})

POST 请求

# JSON 请求
r = requests.post(url, json={"name": "test"})

# 表单数据
r = requests.post(url, data={"key": "value"})

# 多部分文件上传
r = requests.post(url, files={"file": open("test.txt", "rb")})

响应处理

r = requests.get(url)

r.status_code    # 状态码
r.text           # 文本内容
r.json()         # JSON 解析
r.headers        # 响应头
r.cookies       # cookies

# 常用判断
r.raise_for_status()  # 状态码异常时抛出

Session 会话

# 保持 cookie
session = requests.Session()
session.headers.update({"Authorization": "token xxx"})
r = session.get(url)  # 自动携带认证

超时与重试

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

session = requests.Session()
retry = Retry(total=3, backoff_factor=0.5)
adapter = HTTPAdapter(max_retries=retry)
session.mount("http://", adapter)
session.mount("https://", adapter)

r = requests.get(url, timeout=10)

响应处理

import requests

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

# 状态码
if response.status_code == 200:
    print("请求成功")
elif response.status_code == 404:
    print("资源未找到")

# 响应内容
print(f"文本: {response.text[:100]}...")      # 文本格式
print(f"JSON: {response.json()}")             # JSON 格式
print(f"二进制: {response.content[:100]}")    # 二进制格式

# 响应头
print(f"Content-Type: {response.headers['Content-Type']}")

# Cookies
print(f"Cookies: {response.cookies}")

会话管理

import requests

# 创建会话
session = requests.Session()

# 设置会话参数
session.headers.update({'User-Agent': 'MyApp/1.0'})
session.auth = ('username', 'password')

# 使用会话发送请求
response1 = session.get('https://api.github.com/user')
response2 = session.get('https://api.github.com/user/repos')

# 会话会自动保持 Cookies
print(f"Cookies: {session.cookies}")

# 关闭会话
session.close()

超时和重试

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

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

# 设置重试策略
session = requests.Session()
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
session.mount("http://", adapter)

response = session.get('https://api.github.com')

文件上传

import requests

# 上传文件
files = {
    'file': ('example.txt', open('example.txt', 'rb'), 'text/plain')
}
response = requests.post('https://httpbin.org/post', files=files)
print(response.json())

# 上传多个文件
files = [
    ('files', ('file1.txt', open('file1.txt', 'rb'))),
    ('files', ('file2.txt', open('file2.txt', 'rb')))
]
response = requests.post('https://httpbin.org/post', files=files)

文件下载

import requests

# 下载小文件
response = requests.get('https://example.com/file.txt')
with open('downloaded.txt', 'wb') as f:
    f.write(response.content)

# 下载大文件（流式下载）
response = requests.get('https://example.com/large-file.zip', stream=True)
with open('large-file.zip', 'wb') as f:
    for chunk in response.iter_content(chunk_size=8192):
        if chunk:
            f.write(chunk)

# 显示下载进度
response = requests.get('https://example.com/large-file.zip', stream=True)
total_size = int(response.headers.get('content-length', 0))
downloaded = 0

with open('large-file.zip', 'wb') as f:
    for chunk in response.iter_content(chunk_size=8192):
        if chunk:
            f.write(chunk)
            downloaded += len(chunk)
            progress = (downloaded / total_size) * 100
            print(f"\r下载进度: {progress:.2f}%", end='')

代理设置

import requests

# 设置代理
proxies = {
    'http': 'http://proxy.example.com:8080',
    'https': 'https://proxy.example.com:8080'
}

response = requests.get('https://api.github.com', proxies=proxies)

# 使用 SOCKS 代理（需要安装 requests[socks]）
proxies = {
    'http': 'socks5://user:pass@proxy.example.com:1080',
    'https': 'socks5://user:pass@proxy.example.com:1080'
}

response = requests.get('https://api.github.com', proxies=proxies)

SSL 证书验证

import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning

# 禁用 SSL 警告
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

# 禁用 SSL 验证（不推荐用于生产环境）
response = requests.get('https://example.com', verify=False)

# 使用自定义证书
response = requests.get('https://example.com', cert=('/path/to/client.cert', '/path/to/client.key'))

异常处理

import requests
from requests.exceptions import RequestException

try:
    response = requests.get('https://api.github.com', timeout=5)
    response.raise_for_status()  # 检查 HTTP 错误
    print(response.json())
except requests.Timeout:
    print("请求超时")
except requests.ConnectionError:
    print("连接错误")
except requests.HTTPError as e:
    print(f"HTTP 错误: {e}")
except RequestException as e:
    print(f"请求异常: {e}")