1. 什么是 Requests? #
Requests 是一个优雅而简单的 Python HTTP 库,专门为人类设计。它让 HTTP 请求变得非常简单,是 Python 中最流行的 HTTP 客户端库。
为什么要用 Requests?
想象一下,你需要从网页获取数据、调用 API、下载文件,或者上传数据到服务器。手动处理 HTTP 请求很复杂,而 Requests 提供了简单直观的 API,让你只需几行代码就能完成这些任务。
Requests 的优势:
- 简单易用:API 设计非常直观,几行代码就能完成复杂的 HTTP 请求
- 自动处理:自动处理编码、Cookie、连接保持等细节
- 功能全面:支持所有 HTTP 方法、认证、代理等常用功能
- 异常完善:提供详细的错误信息和异常处理
- 文档完善:拥有详细的中英文文档和丰富的示例
前置知识补充:
- HTTP(超文本传输协议):HTTP 是互联网上应用最广泛的网络协议,用于从服务器获取网页、数据等资源。HTTP 定义了客户端(如浏览器)和服务器之间的通信规则。
- URL(统一资源定位符):URL 是互联网上资源的地址,例如
https://www.example.com/page。URL 包含协议(http/https)、域名、路径等信息。- GET 和 POST 请求:GET 请求用于获取数据(如查看网页),POST 请求用于提交数据(如填写表单)。还有 PUT、DELETE、PATCH 等其他请求方法。
- 状态码:HTTP 响应包含状态码,如 200(成功)、404(未找到)、500(服务器错误)等。
- Cookie:Cookie 是服务器存储在客户端的小型数据文件,用于保持用户会话状态(如登录状态)。
- JSON(JavaScript 对象表示法):JSON 是一种常用的数据交换格式,类似于 Python 的字典和列表。
2. 环境准备 #
在学习 Requests 之前,需要确保你的 Python 环境已经准备好。Requests 支持 Python 3.7 及以上版本。
2.1 检查 Python 版本 #
在安装之前,先检查一下你的 Python 版本是否符合要求。
# Windows PowerShell:查看 Python 版本
python --version# macOS 终端:查看 Python 版本
python3 --version2.2 安装 Requests #
Requests 可以直接通过 pip 安装。安装过程会自动下载所需的依赖包。
# 说明:Windows PowerShell 安装 Requests
# 先升级 pip 到最新版本,确保能正常安装依赖
python -m pip install --upgrade pip
# 安装 requests 库
python -m pip install requests# 说明:macOS / Linux 终端安装 Requests
# 先升级 pip 到最新版本
python3 -m pip install --upgrade pip
# 安装 requests 库
python3 -m pip install requests网络加速提示:如果从 PyPI 下载较慢,可以使用国内镜像:
- Windows:
python -m pip install requests -i https://pypi.tuna.tsinghua.edu.cn/simple- macOS:
python3 -m pip install requests -i https://pypi.tuna.tsinghua.edu.cn/simple
安装问题提示:
- 如果安装失败,可能是因为网络问题。可以尝试使用镜像源,或者检查网络连接。
- macOS 用户如果遇到权限问题,可以在命令前加
sudo(不推荐),或者使用虚拟环境。
2.3 验证安装 #
安装完成后,验证一下是否安装成功。
# -*- coding: utf-8 -*-
# 说明:验证 Requests 是否安装成功
# 说明:导入 requests 模块
# requests 是用于发送 HTTP 请求的核心库
import requests
# 说明:打印 Requests 版本信息
# __version__ 属性返回库的版本号
print(f"Requests 版本:{requests.__version__}")
# 说明:测试发送一个简单的 GET 请求
# get() 方法发送 GET 请求
# httpbin.org 是一个用于测试 HTTP 请求的网站
response = requests.get("https://httpbin.org/get")
# 说明:检查响应状态码
# status_code 属性返回 HTTP 状态码
# 200 表示请求成功
if response.status_code == 200:
print("✓ Requests 安装成功!")
print(f"✓ 测试请求成功,状态码:{response.status_code}")
else:
print(f"✗ 测试请求失败,状态码:{response.status_code}")
# 说明:如果没有报错并输出了成功信息,说明安装成功3. 核心概念 #
在使用 Requests 之前,需要理解几个核心概念。这些概念是理解 Requests 的基础。
3.1 请求(Request) #
请求是客户端向服务器发送的消息,包含 URL、请求方法、请求头、请求体等信息。Requests 提供了 get()、post()、put()、delete() 等方法来发送不同类型的请求。
3.2 响应(Response) #
响应是服务器返回的消息,包含状态码、响应头、响应体等信息。Requests 的请求方法返回一个 Response 对象,你可以通过这个对象获取响应的内容。
3.3 会话(Session) #
会话用于保持多个请求之间的状态,如 Cookie、连接池等。使用 Session 可以提高性能,特别是在需要发送多个请求的场景下。
4. 快速体验:发送第一个请求 #
让我们通过一个最简单的例子来快速体验 Requests 的强大功能。
4.1 发送 GET 请求 #
| 方法/属性 | 功能说明 | 常用参数 | 返回值 |
|---|---|---|---|
requests.get() |
发送 GET 请求 | url:请求的 URL,params:URL 参数(可选),headers:请求头(可选) |
Response 对象 |
.status_code |
获取响应状态码 | 无(属性) | 整数(如 200、404) |
.text |
获取响应文本内容 | 无(属性) | 字符串 |
.json() |
解析 JSON 响应 | 无(方法) | 字典或列表 |
# -*- coding: utf-8 -*-
# 说明:快速体验 Requests - 发送第一个 GET 请求
# 说明:导入 requests 模块
# requests 是用于发送 HTTP 请求的核心库
import requests
# 说明:发送最简单的 GET 请求
# get() 方法发送 GET 请求
# 参数 url 是请求的网址
# httpbin.org/get 是一个测试接口,会返回请求信息
response = requests.get("https://httpbin.org/get")
# 说明:检查响应状态码
# status_code 属性返回 HTTP 状态码
# 200 表示请求成功,404 表示未找到,500 表示服务器错误
print(f"状态码:{response.status_code}")
# 说明:获取响应文本内容
# text 属性返回响应的文本内容(字符串格式)
# 这是响应体(response body)的内容
print(f"\n响应内容:\n{response.text}")
# 说明:如果响应是 JSON 格式,可以使用 json() 方法解析
# json() 方法将 JSON 格式的文本解析为 Python 字典或列表
json_data = response.json()
print(f"\n解析后的 JSON 数据:")
print(json_data)
# 说明:访问 JSON 数据中的字段
# 解析后可以像访问字典一样访问数据
if "url" in json_data:
print(f"\n请求的 URL:{json_data['url']}")5. 基本使用:各种 HTTP 请求方法 #
Requests 支持所有常用的 HTTP 请求方法。本节介绍如何发送不同类型的请求。
5.1 GET 请求:获取数据 #
GET 请求用于从服务器获取数据,是最常用的请求方法。
| 方法 | 功能说明 | 常用参数 | 返回值 |
|---|---|---|---|
requests.get() |
发送 GET 请求 | url:请求的 URL,params:URL 参数(字典),headers:请求头(字典) |
Response 对象 |
# -*- coding: utf-8 -*-
# 说明:演示如何发送 GET 请求
# 说明:导入 requests 模块
import requests
# 说明:示例1:最简单的 GET 请求
# 不传任何参数,直接访问 URL
response = requests.get("https://httpbin.org/get")
print("示例1:最简单的 GET 请求")
print(f"状态码:{response.status_code}")
print(f"URL:{response.json()['url']}\n")
# 说明:示例2:带 URL 参数的 GET 请求
# params 参数传递 URL 查询参数
# 这些参数会被自动添加到 URL 后面,如 ?key1=value1&key2=value2
params = {
"key1": "value1",
"key2": "value2"
}
# 说明:发送带参数的 GET 请求
# params 参数是一个字典,会被自动转换为 URL 查询字符串
response = requests.get("https://httpbin.org/get", params=params)
# 说明:查看实际请求的 URL
# url 属性返回实际请求的 URL(包含参数)
print("示例2:带 URL 参数的 GET 请求")
print(f"实际请求 URL:{response.url}")
# 说明:查看服务器收到的参数
# json() 方法解析响应,返回字典
# args 字段包含 URL 参数
json_data = response.json()
print(f"服务器收到的参数:{json_data['args']}\n")
# 说明:示例3:带请求头的 GET 请求
# headers 参数设置请求头
# 请求头包含客户端信息,如浏览器类型、语言偏好等
headers = {
"User-Agent": "MyApp/1.0", # 用户代理(客户端应用名称)
"Accept": "application/json" # 接受的内容类型
}
# 说明:发送带请求头的 GET 请求
response = requests.get("https://httpbin.org/headers", headers=headers)
print("示例3:带请求头的 GET 请求")
print(f"服务器收到的请求头:")
headers_received = response.json()["headers"]
for key, value in headers_received.items():
print(f" {key}:{value}")5.2 POST 请求:提交数据 #
POST 请求用于向服务器提交数据,常用于表单提交、API 调用等场景。
| 方法 | 功能说明 | 常用参数 | 返回值 |
|---|---|---|---|
requests.post() |
发送 POST 请求 | url:请求的 URL,data:表单数据(字典),json:JSON 数据(字典),files:文件(字典) |
Response 对象 |
# -*- coding: utf-8 -*-
# 说明:演示如何发送 POST 请求
# 说明:导入 requests 模块
import requests
# 说明:示例1:发送表单数据(application/x-www-form-urlencoded)
# data 参数传递表单数据
# 这些数据会被编码为表单格式发送
data = {
"username": "john_doe",
"password": "secret123"
}
# 说明:发送 POST 请求
# post() 方法发送 POST 请求
# data 参数传递表单数据
response = requests.post("https://httpbin.org/post", data=data)
print("示例1:发送表单数据")
print(f"状态码:{response.status_code}")
# 说明:查看服务器收到的表单数据
# form 字段包含表单数据
form_data = response.json()["form"]
print(f"服务器收到的表单数据:{form_data}\n")
# 说明:示例2:发送 JSON 数据(application/json)
# json 参数传递 JSON 数据
# Requests 会自动将字典转换为 JSON 格式,并设置正确的请求头
json_data = {
"name": "John Doe",
"email": "john@example.com",
"age": 30
}
# 说明:发送 JSON 数据
# json 参数传递 JSON 数据
# Requests 会自动设置 Content-Type 为 application/json
response = requests.post("https://httpbin.org/post", json=json_data)
print("示例2:发送 JSON 数据")
print(f"状态码:{response.status_code}")
# 说明:查看服务器收到的 JSON 数据
# json 字段包含 JSON 数据
received_json = response.json()["json"]
print(f"服务器收到的 JSON 数据:{received_json}")
# 说明:查看请求头
# headers 字段包含请求头信息
# 可以看到 Content-Type 被自动设置为 application/json
content_type = response.json()["headers"].get("Content-Type", "")
print(f"Content-Type:{content_type}\n")
# 说明:示例3:上传文件
# files 参数传递文件
# 需要先准备一个测试文件
import os
# 说明:创建一个测试文件
test_file = "test.txt"
with open(test_file, "w", encoding="utf-8") as f:
f.write("这是测试文件内容")
# 说明:准备文件上传
# files 参数是一个字典,键是字段名,值是文件对象或元组
# 元组格式:(文件名, 文件对象, 内容类型)
files = {
"file": (test_file, open(test_file, "rb"), "text/plain")
}
# 说明:发送文件上传请求
response = requests.post("https://httpbin.org/post", files=files)
# 说明:关闭文件对象
files["file"][1].close()
print("示例3:上传文件")
print(f"状态码:{response.status_code}")
# 说明:查看服务器收到的文件信息
file_info = response.json()["files"]
print(f"上传的文件名:{list(file_info.keys())[0]}")
# 说明:清理测试文件
if os.path.exists(test_file):
os.remove(test_file)5.3 其他 HTTP 方法 #
除了 GET 和 POST,Requests 还支持其他 HTTP 方法,如 PUT、DELETE、PATCH 等。
| 方法 | 功能说明 | 常用参数 | 返回值 |
|---|---|---|---|
requests.put() |
发送 PUT 请求 | url:请求的 URL,data:数据 |
Response 对象 |
requests.delete() |
发送 DELETE 请求 | url:请求的 URL |
Response 对象 |
requests.patch() |
发送 PATCH 请求 | url:请求的 URL,data:数据 |
Response 对象 |
requests.head() |
发送 HEAD 请求 | url:请求的 URL |
Response 对象 |
# -*- coding: utf-8 -*-
# 说明:演示其他 HTTP 请求方法
# 说明:导入 requests 模块
import requests
# 说明:示例1:PUT 请求(更新资源)
# put() 方法发送 PUT 请求
# 通常用于更新服务器上的资源
data = {"key": "updated_value"}
# 说明:发送 PUT 请求
response = requests.put("https://httpbin.org/put", data=data)
print("示例1:PUT 请求")
print(f"状态码:{response.status_code}")
# 说明:示例2:DELETE 请求(删除资源)
# delete() 方法发送 DELETE 请求
# 通常用于删除服务器上的资源
response = requests.delete("https://httpbin.org/delete")
print("\n示例2:DELETE 请求")
print(f"状态码:{response.status_code}")
# 说明:示例3:PATCH 请求(部分更新资源)
# patch() 方法发送 PATCH 请求
# 通常用于部分更新资源(只更新部分字段)
data = {"key": "partially_updated"}
# 说明:发送 PATCH 请求
response = requests.patch("https://httpbin.org/patch", data=data)
print("\n示例3:PATCH 请求")
print(f"状态码:{response.status_code}")
# 说明:示例4:HEAD 请求(只获取响应头)
# head() 方法发送 HEAD 请求
# 只返回响应头,不返回响应体,用于检查资源是否存在
response = requests.head("https://httpbin.org/get")
print("\n示例4:HEAD 请求")
print(f"状态码:{response.status_code}")
# 说明:HEAD 请求不返回响应体
# content 属性返回响应体内容,HEAD 请求的响应体为空
print(f"响应体长度:{len(response.content)}")
# 说明:但可以获取响应头信息
# headers 属性返回响应头字典
print(f"响应头数量:{len(response.headers)}")6. 响应处理:解析和检查响应 #
发送请求后,需要处理服务器的响应。本节介绍如何解析和检查响应内容。
6.1 检查响应状态 #
本代码块中使用的 Requests 方法/属性:
| 方法/属性 | 功能说明 | 常用参数 | 返回值 |
|---|---|---|---|
.status_code |
获取响应状态码 | 无(属性) | 整数(如 200、404) |
.reason |
获取状态码描述 | 无(属性) | 字符串(如 "OK"、"Not Found") |
.headers |
获取响应头 | 无(属性) | 字典 |
.url |
获取最终请求的 URL | 无(属性) | 字符串 |
.cookies |
获取响应 Cookie | 无(属性) | RequestsCookieJar 对象 |
.raise_for_status() |
如果状态码不是 200-299,抛出异常 | 无(方法) | 无 |
# -*- coding: utf-8 -*-
# 说明:演示如何检查响应状态
# 说明:导入 requests 模块
import requests
# 说明:发送一个 GET 请求
response = requests.get("https://httpbin.org/get")
# 说明:检查响应状态码
# status_code 属性返回 HTTP 状态码
# 200-299 表示成功,300-399 表示重定向,400-499 表示客户端错误,500-599 表示服务器错误
print(f"状态码:{response.status_code}")
# 说明:获取状态码描述
# reason 属性返回状态码的文字描述
# 200 对应 "OK",404 对应 "Not Found"
print(f"状态描述:{response.reason}")
# 说明:检查请求是否成功
# status_code 在 200-299 范围内表示成功
if 200 <= response.status_code < 300:
print("✓ 请求成功")
else:
print("✗ 请求失败")
# 说明:使用 raise_for_status() 检查状态
# raise_for_status() 方法如果状态码不是 200-299,会抛出 HTTPError 异常
try:
response.raise_for_status()
print("✓ 响应状态正常")
except requests.exceptions.HTTPError as e:
print(f"✗ HTTP 错误:{e}")
# 说明:获取响应头信息
# headers 属性返回响应头字典
# 响应头包含服务器返回的元数据,如内容类型、服务器信息等
print(f"\n响应头:")
for key, value in response.headers.items():
print(f" {key}:{value}")
# 说明:获取最终请求的 URL
# url 属性返回实际请求的 URL(可能经过重定向)
print(f"\n最终请求 URL:{response.url}")
# 说明:获取响应 Cookie
# cookies 属性返回 Cookie 对象
print(f"\nCookie:{response.cookies}")
# 说明:示例:处理错误状态码
print("\n" + "="*50)
print("处理错误状态码示例:")
# 说明:发送一个会返回 404 错误的请求
# status/404 端点会返回 404 状态码
response = requests.get("https://httpbin.org/status/404")
print(f"状态码:{response.status_code}")
# 说明:使用 try-except 处理错误
try:
# 说明:如果状态码不是 200-299,会抛出异常
response.raise_for_status()
except requests.exceptions.HTTPError as e:
print(f"捕获到 HTTP 错误:{e}")
print(f"错误状态码:{e.response.status_code}")6.2 解析响应内容 #
本代码块中使用的 Requests 方法/属性:
| 方法/属性 | 功能说明 | 常用参数 | 返回值 |
|---|---|---|---|
.text |
获取响应文本内容 | 无(属性) | 字符串 |
.content |
获取响应二进制内容 | 无(属性) | 字节串 |
.json() |
解析 JSON 响应 | 无(方法) | 字典或列表 |
.encoding |
获取或设置响应编码 | 无(属性),可赋值 | 字符串 |
# -*- coding: utf-8 -*-
# 说明:演示如何解析响应内容
# 说明:导入 requests 模块
import requests
# 说明:示例1:获取文本内容
# 发送一个 GET 请求
response = requests.get("https://httpbin.org/json")
# 说明:获取响应文本内容
# text 属性返回响应体作为字符串
# Requests 会自动根据响应头检测编码
text_content = response.text
print("示例1:文本内容")
print(f"内容类型:{type(text_content)}")
print(f"内容长度:{len(text_content)} 字符")
print(f"内容(前100字符):{text_content[:100]}\n")
# 说明:示例2:获取二进制内容
# content 属性返回响应体作为字节串(bytes)
# 适用于图片、视频等二进制文件
binary_content = response.content
print("示例2:二进制内容")
print(f"内容类型:{type(binary_content)}")
print(f"内容长度:{len(binary_content)} 字节\n")
# 说明:示例3:解析 JSON 响应
# json() 方法将 JSON 格式的文本解析为 Python 字典或列表
# 如果响应不是有效的 JSON,会抛出 ValueError 异常
try:
# 说明:解析 JSON 数据
json_data = response.json()
print("示例3:JSON 解析")
print(f"JSON 数据类型:{type(json_data)}")
print(f"JSON 数据:{json_data}")
# 说明:访问 JSON 数据中的字段
# 解析后可以像访问字典一样访问数据
if isinstance(json_data, dict):
print(f"\n访问 JSON 字段:")
for key, value in json_data.items():
print(f" {key}:{value}")
except ValueError:
# 说明:如果响应不是有效的 JSON,捕获异常
print("响应不是有效的 JSON 格式")
# 说明:示例4:处理编码问题
print("\n示例4:编码处理")
response = requests.get("https://httpbin.org/html")
# 说明:查看自动检测的编码
# encoding 属性返回 Requests 自动检测的编码
print(f"自动检测的编码:{response.encoding}")
# 说明:手动设置编码
# encoding 属性可以手动设置,如果自动检测失败
response.encoding = "utf-8"
print(f"设置后的编码:{response.encoding}")
# 说明:如果编码检测失败,可以手动解码二进制内容
if response.encoding is None:
# 说明:尝试使用 UTF-8 解码
try:
text = response.content.decode("utf-8")
print("使用 UTF-8 解码成功")
except UnicodeDecodeError:
# 说明:如果 UTF-8 解码失败,尝试其他编码
text = response.content.decode("latin-1")
print("使用 latin-1 解码成功")7. 请求参数详解:设置请求头、Cookie 和超时 #
在实际应用中,经常需要设置请求头、Cookie 和超时时间。本节介绍如何设置这些参数。
7.1 设置请求头 #
| 方法 | 功能说明 | 常用参数 | 返回值 |
|---|---|---|---|
requests.get() |
发送 GET 请求 | url:请求的 URL,headers:请求头(字典) |
Response 对象 |
# -*- coding: utf-8 -*-
# 说明:演示如何设置请求头
# 说明:导入 requests 模块
import requests
# 说明:自定义请求头
# headers 参数是一个字典,包含请求头信息
# 请求头用于告诉服务器客户端的类型、接受的内容类型等信息
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
# User-Agent 告诉服务器客户端类型(如浏览器、应用程序等)
"Accept": "application/json",
# Accept 告诉服务器客户端接受的内容类型
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
# Accept-Language 告诉服务器客户端偏好的语言
"Referer": "https://example.com",
# Referer 告诉服务器请求来源页面
"Authorization": "Bearer your_token_here"
# Authorization 用于身份认证,Bearer token 是常见的认证方式
}
# 说明:发送带自定义请求头的 GET 请求
# headers 参数设置请求头
response = requests.get(
"https://httpbin.org/headers",
headers=headers
)
print("自定义请求头:")
print(f"状态码:{response.status_code}")
# 说明:查看服务器收到的请求头
# headers 字段包含服务器收到的请求头
received_headers = response.json()["headers"]
print("\n服务器收到的请求头:")
for key, value in received_headers.items():
# 说明:只显示我们设置的请求头
if key.lower() in ["user-agent", "accept", "accept-language", "referer", "authorization"]:
print(f" {key}:{value}")7.2 Cookie 处理 #
| 方法 | 功能说明 | 常用参数 | 返回值 |
|---|---|---|---|
requests.get() |
发送 GET 请求 | url:请求的 URL,cookies:Cookie(字典) |
Response 对象 |
.cookies |
获取响应 Cookie | 无(属性) | RequestsCookieJar 对象 |
.get() |
从 Cookie 中获取值 | name:Cookie 名称 |
Cookie 值 |
# -*- coding: utf-8 -*-
# 说明:演示如何处理 Cookie
# 说明:导入 requests 模块
import requests
# 说明:示例1:发送 Cookie
# cookies 参数传递 Cookie
# Cookie 是服务器存储在客户端的小型数据,用于保持会话状态
cookies = {
"session_id": "abc123",
# session_id 是会话 ID,用于识别用户会话
"user_preference": "dark_mode"
# user_preference 是用户偏好设置
}
# 说明:发送带 Cookie 的 GET 请求
# cookies 参数设置 Cookie
response = requests.get(
"https://httpbin.org/cookies",
cookies=cookies
)
print("示例1:发送 Cookie")
print(f"状态码:{response.status_code}")
# 说明:查看服务器收到的 Cookie
# cookies 字段包含服务器收到的 Cookie
received_cookies = response.json()["cookies"]
print(f"服务器收到的 Cookie:{received_cookies}\n")
# 说明:示例2:接收和保存 Cookie
# 某些网站会在响应中设置 Cookie
# httpbin.org/cookies/set 端点会设置 Cookie
response = requests.get("https://httpbin.org/cookies/set?name=value&age=25")
print("示例2:接收 Cookie")
print(f"状态码:{response.status_code}")
# 说明:获取响应中的 Cookie
# cookies 属性返回 Cookie 对象
print(f"响应 Cookie:{response.cookies}")
# 说明:获取特定的 Cookie 值
# get() 方法从 Cookie 中获取特定名称的值
cookie_value = response.cookies.get("name")
print(f"name Cookie 的值:{cookie_value}")
# 说明:遍历所有 Cookie
print("\n所有 Cookie:")
for cookie in response.cookies:
print(f" {cookie.name}:{cookie.value}")7.3 超时设置 #
| 方法 | 功能说明 | 常用参数 | 返回值 |
|---|---|---|---|
requests.get() |
发送 GET 请求 | url:请求的 URL,timeout:超时时间(秒) |
Response 对象 |
# -*- coding: utf-8 -*-
# 说明:演示如何设置超时
# 说明:导入 requests 模块和异常类
import requests
from requests.exceptions import Timeout, ConnectTimeout, ReadTimeout
# 说明:示例1:设置总超时时间
# timeout 参数设置超时时间(秒)
# 如果请求在指定时间内没有完成,会抛出 Timeout 异常
try:
# 说明:设置超时时间为 3 秒
# timeout=3 表示总超时时间为 3 秒(连接 + 读取)
response = requests.get(
"https://httpbin.org/delay/2", # 这个端点会延迟 2 秒响应
timeout=3 # 3 秒超时
)
print("示例1:超时设置(成功)")
print(f"状态码:{response.status_code}")
except Timeout:
# 说明:如果请求超时,捕获异常
print("请求超时!")
# 说明:示例2:分别设置连接超时和读取超时
# timeout 可以是元组 (连接超时, 读取超时)
# 连接超时是建立连接的最大时间,读取超时是等待响应的最大时间
try:
# 说明:设置连接超时为 1 秒,读取超时为 3 秒
response = requests.get(
"https://httpbin.org/delay/2", # 延迟 2 秒
timeout=(1, 3) # (连接超时, 读取超时)
)
print("\n示例2:分别设置连接和读取超时(成功)")
print(f"状态码:{response.status_code}")
except ConnectTimeout:
# 说明:连接超时异常
print("连接超时!")
except ReadTimeout:
# 说明:读取超时异常
print("读取超时!")
except Timeout:
# 说明:总超时异常
print("请求超时!")
# 说明:示例3:处理超时情况
print("\n示例3:处理超时情况")
try:
# 说明:设置一个很短的超时时间(1 秒)
# delay/5 端点会延迟 5 秒,所以一定会超时
response = requests.get(
"https://httpbin.org/delay/5",
timeout=1 # 1 秒超时,但请求需要 5 秒
)
except Timeout as e:
# 说明:捕获超时异常
print(f"请求超时:{e}")
print("提示:服务器响应太慢,请求已取消")
# 说明:最佳实践:总是设置超时时间
# 不设置超时可能导致程序挂起
print("\n最佳实践:总是设置超时时间")
response = requests.get("https://httpbin.org/get", timeout=10)
print(f"请求成功,状态码:{response.status_code}")
print("提示:在实际应用中,建议始终设置合理的超时时间")8. 会话管理:使用 Session 保持状态 #
如果需要发送多个请求并保持状态(如 Cookie),可以使用 Session。Session 还可以复用连接,提高性能。
8.1 使用 Session 保持 Cookie #
| 方法/类 | 功能说明 | 常用参数 | 返回值 |
|---|---|---|---|
requests.Session() |
创建会话对象 | 无 | Session 对象 |
.headers.update() |
更新会话级别的请求头 | headers:请求头(字典) |
无 |
.get() |
发送 GET 请求 | url:请求的 URL |
Response 对象 |
.cookies |
获取会话 Cookie | 无(属性) | RequestsCookieJar 对象 |
.close() |
关闭会话 | 无 | 无 |
# -*- coding: utf-8 -*-
# 说明:演示如何使用 Session 保持状态
# 说明:导入 requests 模块
import requests
# 说明:方法1:使用 with 语句(推荐)
# with 语句确保会话在使用后自动关闭
# Session 会自动保持多个请求之间的 Cookie 和连接
with requests.Session() as session:
# 说明:设置会话级别的请求头
# headers.update() 方法更新请求头
# 这些请求头会应用到所有使用该会话的请求
session.headers.update({
"User-Agent": "MyApp/1.0",
"Accept": "application/json"
})
# 说明:第一次请求,设置 Cookie
# httpbin.org/cookies/set 端点会设置 Cookie
response1 = session.get("https://httpbin.org/cookies/set/session/abc123")
print("第一次请求:")
print(f"状态码:{response1.status_code}")
print(f"会话 Cookie:{session.cookies}")
# 说明:第二次请求,自动携带 Cookie
# 由于使用了同一个 Session,Cookie 会自动携带
response2 = session.get("https://httpbin.org/cookies")
print("\n第二次请求:")
print(f"状态码:{response2.status_code}")
# 说明:查看服务器收到的 Cookie
cookies_received = response2.json()["cookies"]
print(f"服务器收到的 Cookie:{cookies_received}")
# 说明:验证 Cookie 是否保持
if "session" in cookies_received:
print("✓ Cookie 已成功保持")
# 说明:方法2:手动管理会话
# 如果不使用 with 语句,需要手动关闭会话
print("\n" + "="*50)
print("方法2:手动管理会话")
# 说明:创建会话对象
session = requests.Session()
# 说明:设置请求头
session.headers.update({
"User-Agent": "MyApp/1.0",
"Accept": "application/json"
})
try:
# 说明:发送请求
response = session.get("https://httpbin.org/get")
print(f"请求成功,状态码:{response.status_code}")
finally:
# 说明:使用完后关闭会话
# close() 方法关闭会话,释放资源
session.close()
print("会话已关闭")
# 说明:提示
print("\n提示:")
print(" - 使用 with 语句可以自动管理会话的关闭")
print(" - Session 可以复用连接,提高性能")
print(" - Session 会自动保持 Cookie,方便处理登录等场景")9. 文件下载:下载文件 #
Requests 可以下载文件,包括文本文件和二进制文件(如图片、视频)。
9.1 下载文件 #
| 方法/属性 | 功能说明 | 常用参数 | 返回值 |
|---|---|---|---|
requests.get() |
发送 GET 请求 | url:请求的 URL,stream:是否流式下载(True/False) |
Response 对象 |
.iter_content() |
迭代下载内容块 | chunk_size:块大小(字节) |
迭代器 |
.headers.get() |
获取响应头 | key:响应头名称 |
响应头值 |
.raise_for_status() |
检查响应状态 | 无 | 无 |
# -*- coding: utf-8 -*-
# 说明:演示如何下载文件
# 说明:导入 requests 模块和 os 模块
import requests
import os
def download_file(url: str, filename: str = None):
"""
下载文件
参数:
url (str): 文件 URL
filename (str): 保存的文件名(可选,如果不指定会从 URL 提取)
"""
# 说明:如果没有指定文件名,从 URL 提取
if filename is None:
# 说明:从 URL 提取文件名
# 使用 URL 的最后一部分作为文件名
filename = url.split("/")[-1] or "download"
try:
# 说明:发送 GET 请求,启用流式下载
# stream=True 表示流式下载,不会一次性加载整个文件到内存
# 这对于大文件很重要,可以避免内存溢出
response = requests.get(url, stream=True)
# 说明:检查响应状态
# raise_for_status() 如果状态码不是 200-299,会抛出异常
response.raise_for_status()
# 说明:获取文件大小(如果服务器提供)
# content-length 响应头包含文件大小(字节)
total_size = int(response.headers.get("content-length", 0))
# 说明:打开文件准备写入
# "wb" 表示以二进制写入模式打开文件
with open(filename, "wb") as f:
# 说明:记录已下载的大小
downloaded_size = 0
# 说明:分块下载文件
# iter_content() 方法按块迭代响应内容
# chunk_size=8192 表示每次读取 8KB(8192 字节)
for chunk in response.iter_content(chunk_size=8192):
# 说明:过滤掉空的块(keep-alive 连接)
if chunk:
# 说明:写入文件
f.write(chunk)
# 说明:更新已下载大小
downloaded_size += len(chunk)
# 说明:显示下载进度(如果知道总大小)
if total_size > 0:
# 说明:计算下载百分比
percent = (downloaded_size / total_size) * 100
# 说明:使用 \r 在同一行更新进度
print(f"\r下载进度: {percent:.1f}% ({downloaded_size}/{total_size} 字节)", end="", flush=True)
# 说明:下载完成,换行
print(f"\n文件已下载: {filename}")
return filename
except requests.exceptions.RequestException as e:
# 说明:如果下载失败,打印错误信息
print(f"下载失败: {e}")
return None
# 说明:示例:下载图片
print("下载文件示例:")
url = "https://httpbin.org/image/jpeg"
filename = "test_image.jpg"
# 说明:调用下载函数
result = download_file(url, filename)
if result:
# 说明:检查文件是否存在
if os.path.exists(filename):
file_size = os.path.getsize(filename)
print(f"文件大小: {file_size} 字节")
# 说明:清理测试文件(可选)
# os.remove(filename)
# print("测试文件已删除")10. 错误处理:处理异常情况 #
在实际应用中,网络请求可能会失败。本节介绍如何处理各种异常情况。
10.1 常见异常处理 #
本代码块中使用的 Requests 异常类:
| 异常类 | 说明 | 何时抛出 |
|---|---|---|
requests.exceptions.HTTPError |
HTTP 错误 | 状态码不是 200-299 |
requests.exceptions.ConnectionError |
连接错误 | 无法连接到服务器 |
requests.exceptions.Timeout |
超时错误 | 请求超时 |
requests.exceptions.RequestException |
请求异常 | 所有请求异常的基类 |
# -*- coding: utf-8 -*-
# 说明:演示如何处理请求异常
# 说明:导入 requests 模块和异常类
import requests
from requests.exceptions import (
HTTPError,
ConnectionError,
Timeout,
RequestException
)
def robust_request(url: str, **kwargs):
"""
健壮的请求函数,包含完整的错误处理
参数:
url (str): 请求的 URL
**kwargs: 其他请求参数(如 timeout、headers 等)
返回:
Response 对象或 None(如果请求失败)
"""
try:
# 说明:发送 GET 请求
# get() 方法发送 GET 请求
response = requests.get(url, **kwargs)
# 说明:检查响应状态
# raise_for_status() 如果状态码不是 200-299,会抛出 HTTPError
response.raise_for_status()
# 说明:请求成功,返回响应对象
return response
except HTTPError as e:
# 说明:HTTP 错误(如 404、500 等)
print(f"HTTP 错误: {e}")
# 说明:访问响应对象获取状态码
if hasattr(e, "response") and e.response is not None:
print(f"状态码: {e.response.status_code}")
except ConnectionError as e:
# 说明:连接错误(如网络问题、DNS 解析失败等)
print(f"连接错误: {e}")
print("提示:请检查网络连接或 URL 是否正确")
except Timeout as e:
# 说明:超时错误(请求超过指定时间)
print(f"超时错误: {e}")
print("提示:服务器响应太慢,可以增加超时时间")
except RequestException as e:
# 说明:其他请求异常
print(f"请求异常: {e}")
except Exception as e:
# 说明:未知错误
print(f"未知错误: {e}")
# 说明:请求失败,返回 None
return None
# 说明:示例1:处理 404 错误
print("示例1:处理 404 错误")
response = robust_request("https://httpbin.org/status/404")
if response:
print("请求成功")
else:
print("请求失败\n")
# 说明:示例2:处理连接错误(使用错误的 URL)
print("示例2:处理连接错误")
response = robust_request("https://nonexistent-domain-12345.com", timeout=5)
if response:
print("请求成功")
else:
print("请求失败\n")
# 说明:示例3:处理超时错误
print("示例3:处理超时错误")
response = robust_request(
"https://httpbin.org/delay/10", # 延迟 10 秒
timeout=2 # 2 秒超时
)
if response:
print("请求成功")
else:
print("请求失败\n")
# 说明:示例4:成功请求
print("示例4:成功请求")
response = robust_request("https://httpbin.org/get", timeout=10)
if response:
print("✓ 请求成功")
print(f"状态码: {response.status_code}")
else:
print("请求失败")11. 实战案例:API 客户端 #
下面是一个完整的实战案例,综合运用前面学到的知识,创建一个简单的 API 客户端。
| 方法/属性 | 功能说明 |
|---|---|
requests.Session() |
创建会话 |
.headers.update() |
更新请求头 |
.get() |
发送 GET 请求 |
.post() |
发送 POST 请求 |
.json() |
解析 JSON 响应 |
.raise_for_status() |
检查响应状态 |
.close() |
关闭会话 |
# -*- coding: utf-8 -*-
# 说明:实战案例 - 简单的 API 客户端
# 综合运用前面学到的所有知识,创建一个 API 客户端
# 说明:导入所需的库
import requests
from requests.exceptions import RequestException
class APIClient:
"""
通用的 API 客户端
用于调用 RESTful API,支持 GET、POST、PUT、DELETE 等方法
"""
def __init__(self, base_url: str, api_key: str = None):
"""
初始化 API 客户端
参数:
base_url (str): API 的基础 URL(如 https://api.example.com)
api_key (str): API 密钥(可选,用于身份认证)
"""
# 说明:保存基础 URL
# rstrip('/') 移除末尾的斜杠,避免 URL 拼接问题
self.base_url = base_url.rstrip("/")
# 说明:创建会话对象
# Session 可以保持连接和 Cookie,提高性能
self.session = requests.Session()
# 说明:设置通用请求头
# headers.update() 方法更新请求头
# 这些请求头会应用到所有使用该会话的请求
self.session.headers.update({
"Content-Type": "application/json",
# Content-Type 告诉服务器发送的是 JSON 格式数据
"User-Agent": "APIClient/1.0"
# User-Agent 标识客户端应用
})
# 说明:如果有 API 密钥,添加到请求头
if api_key:
# 说明:Bearer token 是常见的 API 认证方式
self.session.headers.update({
"Authorization": f"Bearer {api_key}"
})
def get(self, endpoint: str, params: dict = None):
"""
发送 GET 请求
参数:
endpoint (str): API 端点(如 /users)
params (dict): URL 参数(可选)
返回:
dict: 解析后的 JSON 响应
"""
# 说明:构建完整 URL
# lstrip('/') 移除开头的斜杠,避免双斜杠
url = f"{self.base_url}/{endpoint.lstrip('/')}"
try:
# 说明:发送 GET 请求
# params 参数传递 URL 查询参数
response = self.session.get(url, params=params)
# 说明:检查响应状态
# raise_for_status() 如果状态码不是 200-299,会抛出异常
response.raise_for_status()
# 说明:解析 JSON 响应
# json() 方法将 JSON 格式的文本解析为 Python 字典
return response.json()
except RequestException as e:
# 说明:如果请求失败,打印错误信息
print(f"GET 请求失败: {e}")
raise
def post(self, endpoint: str, data: dict = None):
"""
发送 POST 请求
参数:
endpoint (str): API 端点
data (dict): 请求体数据(会被转换为 JSON)
返回:
dict: 解析后的 JSON 响应
"""
# 说明:构建完整 URL
url = f"{self.base_url}/{endpoint.lstrip('/')}"
try:
# 说明:发送 POST 请求
# json 参数会自动将字典转换为 JSON 格式
# 并设置 Content-Type 为 application/json
response = self.session.post(url, json=data)
# 说明:检查响应状态
response.raise_for_status()
# 说明:解析 JSON 响应
return response.json()
except RequestException as e:
print(f"POST 请求失败: {e}")
raise
def close(self):
"""
关闭会话
释放资源,关闭连接池
"""
# 说明:关闭会话
# close() 方法关闭会话,释放所有资源
self.session.close()
print("API 客户端已关闭")
# 说明:主程序入口
if __name__ == "__main__":
# 说明:创建 API 客户端实例
# 使用 httpbin.org 作为测试 API
client = APIClient("https://httpbin.org")
try:
# 说明:示例1:发送 GET 请求
print("示例1:GET 请求")
result = client.get("/get", params={"key": "value"})
print(f"响应: {result.get('args', {})}\n")
# 说明:示例2:发送 POST 请求
print("示例2:POST 请求")
data = {"name": "John", "email": "john@example.com"}
result = client.post("/post", data=data)
print(f"响应: {result.get('json', {})}\n")
finally:
# 说明:无论成功还是失败,都要关闭客户端
# finally 语句确保代码总是执行
client.close()12. 常见问题与排查 #
在使用 Requests 时,可能会遇到一些常见问题。本节提供解决方案。
12.1 网络连接问题 #
问题:无法连接到服务器,出现连接错误。
解决方案:
# -*- coding: utf-8 -*-
# 说明:演示如何处理网络连接问题
# 说明:导入 requests 模块和异常类
import requests
from requests.exceptions import ConnectionError
def safe_request(url: str, **kwargs):
"""
安全的请求函数,处理连接问题
参数:
url (str): 请求的 URL
**kwargs: 其他请求参数
"""
try:
# 说明:设置合理的超时时间
# timeout 参数防止请求无限期等待
if "timeout" not in kwargs:
kwargs["timeout"] = 10
# 说明:发送请求
response = requests.get(url, **kwargs)
response.raise_for_status()
return response
except ConnectionError as e:
# 说明:连接错误处理
print(f"连接错误: {e}")
print("可能的原因:")
print(" 1. 网络连接问题(检查网络连接)")
print(" 2. URL 错误(检查 URL 是否正确)")
print(" 3. 服务器不可用(服务器可能已关闭)")
print(" 4. DNS 解析失败(检查域名是否正确)")
return None
except Exception as e:
print(f"其他错误: {e}")
return None
# 说明:测试连接
print("测试网络连接:")
response = safe_request("https://httpbin.org/get")
if response:
print("✓ 连接成功")
else:
print("✗ 连接失败")12.2 编码问题 #
问题:响应内容显示乱码。
解决方案:
# -*- coding: utf-8 -*-
# 说明:演示如何处理编码问题
# 说明:导入 requests 模块
import requests
def handle_encoding(response: requests.Response):
"""
处理响应编码
参数:
response (Response): 响应对象
"""
# 说明:方法1:自动检测编码(默认)
# Requests 会自动根据响应头检测编码
print(f"自动检测的编码: {response.encoding}")
# 说明:如果编码检测失败或乱码,手动设置编码
# 常见的中文编码:utf-8、gbk、gb2312
if response.encoding is None or response.encoding.lower() == "iso-8859-1":
# 说明:尝试使用 UTF-8 编码
response.encoding = "utf-8"
print(f"手动设置编码为: {response.encoding}")
# 说明:获取文本内容
text = response.text
# 说明:如果仍然乱码,尝试其他编码
try:
# 说明:检查文本是否包含中文字符
# 如果包含中文字符,尝试使用 gbk 编码
if "乱码" in text or len(text.encode("utf-8")) != len(text):
# 说明:使用二进制内容重新解码
text = response.content.decode("gbk")
print("使用 GBK 编码解码")
except UnicodeDecodeError:
# 说明:如果解码失败,使用默认编码
text = response.text
return text
# 说明:测试编码处理
print("测试编码处理:")
response = requests.get("https://httpbin.org/html")
text = handle_encoding(response)
print(f"文本长度: {len(text)} 字符")12.3 SSL 证书验证问题 #
问题:出现 SSL 证书验证错误。
解决方案:
# -*- coding: utf-8 -*-
# 说明:演示如何处理 SSL 证书验证问题
# 说明:导入 requests 模块和异常类
import requests
from requests.exceptions import SSLError
# 说明:方法1:验证 SSL 证书(推荐,生产环境)
# verify=True 表示验证 SSL 证书(默认)
try:
response = requests.get("https://httpbin.org/get", verify=True)
print("✓ SSL 证书验证成功")
except SSLError as e:
print(f"✗ SSL 证书验证失败: {e}")
# 说明:方法2:跳过 SSL 证书验证(不推荐,仅用于测试)
# verify=False 表示跳过 SSL 证书验证
# 警告:这会降低安全性,不建议在生产环境使用
print("\n提示:")
print(" - 生产环境应该验证 SSL 证书(verify=True)")
print(" - 只在测试环境或内部网络中使用 verify=False")
print(" - 跳过 SSL 验证会显示警告信息,可以禁用警告:")
print(" import urllib3")
print(" urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)")
# 说明:示例:跳过 SSL 验证(仅用于演示)
# response = requests.get("https://httpbin.org/get", verify=False)