Python Requests库完全指南

1
import requests
2

3
# 基本GET请求
4
response = requests.get('https://api.github.com')
5
print(response.status_code)  # 状态码
6
print(response.text)  # 响应内容（字符串）

1
# 方式1：使用字典
2
params = {'key1': 'value1', 'key2': 'value2'}
3
response = requests.get('https://httpbin.org/get', params=params)
4

5
# 方式2：使用列表（支持重复键）
6
params = [('key', 'value1'), ('key', 'value2')]
7
response = requests.get('https://httpbin.org/get', params=params)
8

9
print(response.url)  # 查看完整URL

1
# 发送表单数据
2
data = {'username': 'user', 'password': 'pass'}
3
response = requests.post('https://httpbin.org/post', data=data)
4

5
# 发送JSON数据
6
json_data = {'name': 'John', 'age': 30}
7
response = requests.post('https://httpbin.org/post', json=json_data)

1
# PUT请求
2
response = requests.put('https://httpbin.org/put', data={'key': 'value'})
3

4
# DELETE请求
5
response = requests.delete('https://httpbin.org/delete')
6

7
# HEAD请求（只获取响应头）
8
response = requests.head('https://httpbin.org/get')
9

10
# OPTIONS请求
11
response = requests.options('https://httpbin.org/get')
12

13
# PATCH请求
14
response = requests.patch('https://httpbin.org/patch', data={'key': 'value'})

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

3
# 文本内容（自动解码）
4
print(response.text)
5

6
# 二进制内容
7
print(response.content)
8

9
# JSON内容（自动解析）
10
data = response.json()
11

12
# 原始响应（需要设置stream=True）
13
response = requests.get('https://api.github.com', stream=True)
14
print(response.raw.read(10))

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

3
# 状态码
4
print(response.status_code)
5

6
# 检查请求是否成功
7
if response.status_code == 200:
8
    print('请求成功')
9

10
# 使用内置状态码常量
11
if response.status_code == requests.codes.ok:
12
    print('请求成功')
13

14
# 自动抛出异常（如果状态码表示错误）
15
response.raise_for_status()

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

3
# 访问响应头（字典形式）
4
print(response.headers)
5
print(response.headers['Content-Type'])
6
print(response.headers.get('content-type'))
7

8
# 响应头不区分大小写
9
print(response.headers['content-type'])
10
print(response.headers['Content-Type'])

1
headers = {
2
    'User-Agent': 'Mozilla/5.0',
3
    'Accept': 'application/json',
4
    'Authorization': 'Bearer token123'
5
}
6

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

1
from requests.auth import HTTPBasicAuth
2

3
# 方式1：使用auth参数
4
response = requests.get(
5
    'https://api.github.com/user',
6
    auth=HTTPBasicAuth('username', 'password')
7
)
8

9
# 方式2：简写形式
10
response = requests.get(
11
    'https://api.github.com/user',
12
    auth=('username', 'password')
13
)

1
# Bearer Token
2
headers = {'Authorization': 'Bearer YOUR_TOKEN'}
3
response = requests.get('https://api.example.com/data', headers=headers)
4

5
# API Key
6
params = {'api_key': 'YOUR_API_KEY'}
7
response = requests.get('https://api.example.com/data', params=params)

1
# 方式1：使用字典
2
cookies = {'session_id': '123456'}
3
response = requests.get('https://httpbin.org/cookies', cookies=cookies)
4

5
# 方式2：使用RequestsCookieJar
6
from requests.cookies import RequestsCookieJar
7

8
jar = RequestsCookieJar()
9
jar.set('cookie_name', 'cookie_value', domain='httpbin.org', path='/cookies')
10
response = requests.get('https://httpbin.org/cookies', cookies=jar)

1
response = requests.get('https://httpbin.org/cookies/set/name/value')
2

3
# 访问cookies
4
print(response.cookies)
5
print(response.cookies['name'])
6

7
# 遍历cookies
8
for cookie in response.cookies:
9
    print(f'{cookie.name}: {cookie.value}')

1
# 上传单个文件
2
files = {'file': open('report.txt', 'rb')}
3
response = requests.post('https://httpbin.org/post', files=files)
4

5
# 指定文件名和内容类型
6
files = {
7
    'file': ('report.pdf', open('report.pdf', 'rb'), 'application/pdf')
8
}
9
response = requests.post('https://httpbin.org/post', files=files)
10

11
# 上传多个文件
12
files = {
13
    'file1': open('file1.txt', 'rb'),
14
    'file2': open('file2.txt', 'rb')
15
}
16
response = requests.post('https://httpbin.org/post', files=files)
17

18
# 同时发送表单数据
19
files = {'file': open('report.txt', 'rb')}
20
data = {'description': 'My report'}
21
response = requests.post('https://httpbin.org/post', files=files, data=data)
展开代码收起代码

1
# 小文件下载
2
response = requests.get('https://example.com/file.pdf')
3
with open('downloaded_file.pdf', 'wb') as f:
4
    f.write(response.content)
5

6
# 大文件流式下载（节省内存）
7
response = requests.get('https://example.com/large_file.zip', stream=True)
8
with open('large_file.zip', 'wb') as f:
9
    for chunk in response.iter_content(chunk_size=8192):
10
        f.write(chunk)
11

12
# 带进度的下载
13
import os
14
response = requests.get('https://example.com/file.zip', stream=True)
15
total_size = int(response.headers.get('content-length', 0))
16
downloaded = 0
17

18
with open('file.zip', 'wb') as f:
19
    for chunk in response.iter_content(chunk_size=8192):
20
        downloaded += len(chunk)
21
        f.write(chunk)
22
        progress = (downloaded / total_size) * 100
23
        print(f'下载进度: {progress:.2f}%')
展开代码收起代码

1
# 连接超时和读取超时（秒）
2
response = requests.get('https://api.github.com', timeout=5)
3

4
# 分别设置连接超时和读取超时
5
response = requests.get('https://api.github.com', timeout=(3, 10))
6

7
# 永久等待（不推荐）
8
response = requests.get('https://api.github.com', timeout=None)

1
from requests.adapters import HTTPAdapter
2
from requests.packages.urllib3.util.retry import Retry
3

4
# 配置重试策略
5
retry_strategy = Retry(
6
    total=3,  # 总重试次数
7
    backoff_factor=1,  # 重试间隔时间因子
8
    status_forcelist=[429, 500, 502, 503, 504],  # 需要重试的状态码
9
    allowed_methods=["HEAD", "GET", "OPTIONS"]  # 允许重试的方法
10
)
11

12
adapter = HTTPAdapter(max_retries=retry_strategy)
13
session = requests.Session()
14
session.mount("http://", adapter)
15
session.mount("https://", adapter)
16

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

1
import requests
2

3
def get_github_user(username):
4
    """获取GitHub用户信息"""
5
    url = f'https://api.github.com/users/{username}'
6
    headers = {'Accept': 'application/vnd.github.v3+json'}
7

8
    try:
9
        response = requests.get(url, headers=headers, timeout=10)
10
        response.raise_for_status()
11
        return response.json()
12
    except requests.exceptions.RequestException as e:
13
        print(f'请求失败: {e}')
14
        return None
15

16
# 使用示例
17
user_data = get_github_user('octocat')
18
if user_data:
19
    print(f"用户名: {user_data['login']}")
20
    print(f"仓库数: {user_data['public_repos']}")

1
import requests
2
from bs4 import BeautifulSoup
3

4
def scrape_website(url):
5
    """简单的网页爬虫"""
6
    headers = {
7
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
8
    }
9

10
    try:
11
        response = requests.get(url, headers=headers, timeout=10)
12
        response.raise_for_status()
13
        response.encoding = response.apparent_encoding
14

15
        soup = BeautifulSoup(response.text, 'html.parser')
16
        title = soup.find('title').text
17

18
        return {
19
            'url': url,
20
            'title': title,
21
            'status_code': response.status_code
22
        }
23
    except Exception as e:
24
        print(f'爬取失败: {e}')
25
        return None
展开代码收起代码

1
class APIClient:
2
    """RESTful API客户端封装"""
3

4
    def __init__(self, base_url, api_key=None):
5
        self.base_url = base_url
6
        self.session = requests.Session()
7

8
        if api_key:
9
            self.session.headers.update({'Authorization': f'Bearer {api_key}'})
10

11
        self.session.headers.update({
12
            'Content-Type': 'application/json',
13
            'Accept': 'application/json'
14
        })
15

16
    def get(self, endpoint, params=None):
17
        """GET请求"""
18
        url = f'{self.base_url}/{endpoint}'
19
        response = self.session.get(url, params=params, timeout=10)
20
        response.raise_for_status()
21
        return response.json()
22

23
    def post(self, endpoint, data=None):
24
        """POST请求"""
25
        url = f'{self.base_url}/{endpoint}'
26
        response = self.session.post(url, json=data, timeout=10)
27
        response.raise_for_status()
28
        return response.json()
29

30
    def put(self, endpoint, data=None):
31
        """PUT请求"""
32
        url = f'{self.base_url}/{endpoint}'
33
        response = self.session.put(url, json=data, timeout=10)
34
        response.raise_for_status()
35
        return response.json()
36

37
    def delete(self, endpoint):
38
        """DELETE请求"""
39
        url = f'{self.base_url}/{endpoint}'
40
        response = self.session.delete(url, timeout=10)
41
        response.raise_for_status()
42
        return response.status_code == 204
43

44
# 使用示例
45
client = APIClient('https://api.example.com', api_key='your_api_key')
46
users = client.get('users', params={'page': 1})
展开代码收起代码

1
# 不推荐：每次创建新连接
2
for i in range(10):
3
    response = requests.get('https://api.github.com')
4

5
# 推荐：复用连接
6
with requests.Session() as session:
7
    for i in range(10):
8
        response = session.get('https://api.github.com')

1
# 推荐
2
response = requests.get('https://api.github.com', timeout=10)
3

4
# 不推荐
5
response = requests.get('https://api.github.com')  # 可能永久阻塞

1
try:
2
    response = requests.get('https://api.github.com', timeout=10)
3
    response.raise_for_status()
4
    data = response.json()
5
except requests.exceptions.RequestException as e:
6
    # 处理所有requests相关异常
7
    print(f'请求失败: {e}')

1
# 推荐：流式下载
2
response = requests.get('https://example.com/large_file.zip', stream=True)
3
with open('file.zip', 'wb') as f:
4
    for chunk in response.iter_content(chunk_size=8192):
5
        f.write(chunk)
6

7
# 不推荐：一次性加载到内存
8
response = requests.get('https://example.com/large_file.zip')
9
with open('file.zip', 'wb') as f:
10
    f.write(response.content)

1
headers = {
2
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
3
}
4
response = requests.get('https://example.com', headers=headers)