request库的详解
安装：
pip install requests
2. 发送GET请求
# 基本GET请求
response = requests.get('https://httpbin.org/get')# 带参数的GET请求
params = {'key1': 'value1', 'key2': 'value2'}
response = requests.get('https://httpbin.org/get', params=params)# 添加请求头
headers = {'User-Agent': 'my-app/1.0.0'}
response = requests.get('https://httpbin.org/get', headers=headers)3. 发送POST请求
# 表单数据
data = {'key1': 'value1', 'key2': 'value2'}
response = requests.post('https://httpbin.org/post', data=data)# JSON数据
json_data = {'key1': 'value1', 'key2': 'value2'}
response = requests.post('https://httpbin.org/post', json=json_data)# 文件上传
files = {'file': open('test.txt', 'rb')}
response = requests.post('https://httpbin.org/post', files=files)4. 其他HTTP方法
# PUT请求
response = requests.put('https://httpbin.org/put', data={'key': 'value'})# DELETE请求
response = requests.delete('https://httpbin.org/delete')# HEAD请求
response = requests.head('https://httpbin.org/get')# OPTIONS请求
response = requests.options('https://httpbin.org/get')响应处理
1. 检查响应状态
response = requests.get('https://httpbin.org/get')# 状态码
print(response.status_code)# 状态码是否成功（200-400范围内）
print(response.ok)# 抛出异常如果请求失败
response.raise_for_status()2. 响应内容
# 文本内容
print(response.text)# 二进制内容
print(response.content)# JSON内容（自动解析）
print(response.json())# 原始响应（用于流式下载）
with open('file.txt', 'wb') as f:for chunk in response.iter_content(chunk_size=1024):f.write(chunk)3. 响应头信息
print(response.headers)
print(response.headers['Content-Type'])高级功能
1. 会话对象（Session）
# 创建会话（保持cookies和连接）
session = requests.Session()# 设置通用头
session.headers.update({'User-Agent': 'my-app/1.0.0'})# 使用会话发送请求
response1 = session.get('https://httpbin.org/cookies/set/sessioncookie/123456789')
response2 = session.get('https://httpbin.org/cookies')2. 超时设置
# 设置超时时间（秒）
response = requests.get('https://httpbin.org/delay/5', timeout=3)3. 代理设置
proxies = {'http': 'http://10.10.1.10:3128','https': 'http://10.10.1.10:1080',
}
response = requests.get('https://httpbin.org/get', proxies=proxies)4. 身份验证
# HTTP基本认证
from requests.auth import HTTPBasicAuth
response = requests.get('https://httpbin.org/basic-auth/user/pass', auth=HTTPBasicAuth('user', 'pass'))# 简化写法
response = requests.get('https://httpbin.org/basic-auth/user/pass', auth=('user', 'pass'))5. SSL证书验证
# 禁用SSL验证（不推荐用于生产环境）
response = requests.get('https://httpbin.org/get', verify=False)# 使用自定义CA证书
response = requests.get('https://httpbin.org/get', verify='/path/to/certfile')6. 重定向处理
# 禁用重定向
response = requests.get('https://httpbin.org/redirect/1', allow_redirects=False)# 查看重定向历史
print(response.history)7. Cookies处理
# 发送cookies
cookies = dict(cookies_are='working')
response = requests.get('https://httpbin.org/cookies', cookies=cookies)# 接收cookies
print(response.cookies)
print(response.cookies['example_cookie_name'])错误处理
try:response = requests.get('https://httpbin.org/status/404')response.raise_for_status() # 如果状态码不是200-400，抛出异常
except requests.exceptions.HTTPError as err:print(f"HTTP错误: {err}")
except requests.exceptions.ConnectionError as err:print(f"连接错误: {err}")
except requests.exceptions.Timeout as err:print(f"超时错误: {err}")
except requests.exceptions.RequestException as err:print(f"其他错误: {err}")实用示例
1. 下载文件
url = 'https://example.com/file.zip'
response = requests.get(url, stream=True)with open('file.zip', 'wb') as f:for chunk in response.iter_content(chunk_size=1024):if chunk:f.write(chunk)2. 流式处理API
import jsonresponse = requests.get('https://api.example.com/stream', stream=True)for line in response.iter_lines():if line:data = json.loads(line)print(data)3. 带进度显示的文件下载
from tqdm import tqdmurl = 'https://example.com/large-file.zip'
response = requests.get(url, stream=True)total_size = int(response.headers.get('content-length', 0))
block_size = 1024 # 1 Kibibytewith open('large-file.zip', 'wb') as file, tqdm(desc='Downloading',total=total_size,unit='iB',unit_scale=True,
) as progress_bar:for data in response.iter_content(block_size):size = file.write(data)progress_bar.update(size)request.form.get('key')
如果键不存在，返回None（或指定的默认值），不会引发异常。
request.args：获取URL中的查询参数（即GET请求的参数）。
request.json：如果请求的Content-Type是application/json，则包含解析后的JSON数据。
request.files：获取上传的文件。
场景1：request.form.get('key')
from flask import Flask, requestapp = Flask(__name__)@app.route('/submit', methods=['POST'])
def handle_form():# 获取表单数据username = request.form.get('username')email = request.form.get('email')return f"用户名: {username}, 邮箱: {email}"
<form action="/submit" method="post"><input type="text" name="username" placeholder="用户名"><input type="email" name="email" placeholder="邮箱"><input type="submit" value提交">
</form>场景2：request.args
例1：基本用法
假设有一个URL：http://example.com/?name=John&age=30from flask import Flask, requestapp = Flask(__name__)@app.route('/')
def index():name = request.args.get('name') # 获取name参数，如果没有则返回Noneage = request.args.get('age') # 获取age参数，如果没有则返回Nonereturn f"Hello {name}, you are {age} years old."场景3：request.json
请求示例：
{"username": "john_doe","email": "john@example.com","password": "securepassword123","age": 25
}@app.route('/api/register', methods=['POST'])
def register_user():# 检查 JSON 数据if not request.is_json:return jsonify({'error': '请求必须是 JSON 格式'}), 400# 获取 JSON 数据data = request.json# 验证必需字段required_fields = ['username', 'email', 'password']for field in required_fields:if field not in data:return jsonify({'error': f'缺少必需字段: {field}'}), 400# 提取数据username = data.get('username')email = data.get('email')password = data.get('password')age = data.get('age') # 可选字段# 验证数据if len(username) < 3:return jsonify({'error': '用户名至少3个字符'}), 400if len(password) < 6:return jsonify({'error': '密码至少6个字符'}), 400# 处理注册逻辑# user_id = create_user(username, email, password, age)return jsonify({'message': '用户注册成功','user': {'username': username,'email': email,'age': age}}), 201场景4：request.files
from flask import request
import os@app.route('/upload', methods=['POST'])
def upload_file():# 获取普通表单字段description = request.form.get('description', '无描述')# 获取上传的文件file = request.files.get('file')if file and file.filename:filename = file.filenamefile.save(os.path.join('uploads', filename))return f"文件 '{filename}' 上传成功！描述: {description}"else:return "没有选择文件"