← 异步编程 aiohttp →

Requests

Requests 是 Python 中最流行的 HTTP 库,简单易用,功能强大。

安装

pip install requests

GET 请求

import requests

# 基本 GET 请求
response = requests.get('https://api.github.com')
print(f"状态码: {response.status_code}")
print(f"内容: {response.text[:100]}...")

# 带参数的 GET 请求
params = {
    'q': 'python',
    'sort': 'stars'
}
response = requests.get('https://api.github.com/search/repositories', params=params)
print(f"URL: {response.url}")
data = response.json()
print(f"找到 {data['total_count']} 个仓库")

# 添加请求头
headers = {
    'User-Agent': 'MyApp/1.0',
    'Authorization': 'Bearer YOUR_TOKEN'
}
response = requests.get('https://api.github.com/user', headers=headers)

POST 请求

import requests

# 发送 JSON 数据
data = {
    'name': '张三',
    'age': 25
}
response = requests.post('https://httpbin.org/post', json=data)
print(response.json())

# 发送表单数据
form_data = {
    'username': 'admin',
    'password': '123456'
}
response = requests.post('https://httpbin.org/post', data=form_data)
print(response.json())

响应处理

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}")
💡 提示:Requests 是处理 HTTP 请求的首选库,简单易用,功能强大。
← 异步编程 aiohttp →