一、什么是 requests？
1.1 一句话定义
requests 是 Python 最流行的 HTTP 客户端库，让你用几行代码就能发送网络请求、获取网页数据、调用 API。
1.2 生活比喻
┌─────────────────────────────────────────────────────────────┐│ ││ 没有 requests（用标准库 urllib）： ││ → 相当于自己造船过河（写一堆底层代码） ││ ││ 有 requests： ││ → 相当于坐渡船过河（一行代码搞定） ││ ││ 浏览器访问网页： ││ → 你打开浏览器，输入网址，看到页面 ││ ││ requests 访问网页： ││ → 你的程序"假装"是浏览器，向服务器要数据 ││ ││ 你的程序（客户端）──请求──→ 服务器（百度/淘宝/API） ││ 你的程序（客户端）←──响应── 服务器（返回HTML/JSON/图片） ││ │└─────────────────────────────────────────────────────────────┘
1.3 requests vs urllib（标准库）
# ============ 用标准库 urllib（麻烦） ============import urllib.requestimport urllib.parseimport jsonurl = "https://api.example.com/data"params = urllib.parse.urlencode({"name": "张三", "age": 25})full_url = f"{url}?{params}"req = urllib.request.Request(full_url)req.add_header("User-Agent", "MyApp/1.0")with urllib.request.urlopen(req, timeout=10) as response: data = json.loads(response.read().decode("utf-8"))# ============ 用 requests（简洁） ============import requestsresponse = requests.get( "https://api.example.com/data", params={"name": "张三", "age": 25}, headers={"User-Agent": "MyApp/1.0"}, timeout=10)data = response.json()# 代码量减少 60%，可读性提升 200%！
1.4 requests 能做什么？
二、安装
# 方法1：pip 安装（最常用）pip install requests# 方法2：在虚拟环境中安装（推荐）python3 -m venv venvsource venv/bin/activate # Linux/macOSpip install requests# 验证安装python3 -c "import requests; print(requests.__version__)"# 输出：2.32.3（或类似版本号）
# 导入import requests# 查看版本print(requests.__version__)
三、第一个请求（GET）
3.1 最简单的 GET 请求
import requests# 发送 GET 请求（相当于浏览器访问这个网址）response = requests.get("https://httpbin.org/get")# response 是一个 Response 对象，包含服务器的所有回复print(type(response)) # <class 'requests.models.Response'>print(response.status_code) # 200（状态码，200表示成功）print(response.text) # 响应体（字符串形式）
📌 httpbin.org 是一个专门用来测试 HTTP 请求的网站，它会把你发送的请求信息原样返回给你。
3.2 获取一个真实网页
import requests# 获取百度首页response = requests.get("https://www.baidu.com")# 查看状态码print(f"状态码：{response.status_code}") # 200# 查看响应内容（HTML）print(response.text[:500]) # 只看前500个字符# 输出类似：<!DOCTYPE html><html><head><title>百度一下...</title>...# 查看编码print(f"编码：{response.encoding}") # utf-8 或 ISO-8859-1# 查看响应头print(f"内容类型：{response.headers['Content-Type']}")# text/html;charset=utf-8# 查看请求的 URLprint(f"最终URL：{response.url}")# 查看耗时print(f"耗时：{response.elapsed.total_seconds():.3f}秒")
3.3 处理中文编码
import requestsresponse = requests.get("https://www.baidu.com")# 有时候 encoding 检测不正确，需要手动设置response.encoding = "utf-8" # 手动指定编码print(response.text) # 现在中文正常显示了# 或者用 response.content（原始字节）自己解码raw_bytes = response.content # bytes 类型text = raw_bytes.decode("utf-8")print(text)
四、Response 对象详解
4.1 常用属性和方法
import requestsresponse = requests.get("https://httpbin.org/get")# ============ 状态码 ============print(response.status_code) # 200print(response.ok) # True（状态码 < 400 就是 True）print(response.reason) # "OK"# ============ 响应内容 ============print(response.text) # 字符串形式（自动解码）print(response.content) # 字节形式（bytes，适合图片/文件）print(response.json()) # 解析为 Python 字典（如果响应是JSON）# ============ 响应头 ============print(response.headers) # 所有响应头（字典）print(response.headers["Content-Type"]) # "application/json"print(response.headers["Server"]) # 服务器软件# ============ 请求信息 ============print(response.url) # 最终请求的URL（可能有重定向）print(response.request) # 原始的 Request 对象print(response.request.headers) # 发送的请求头print(response.request.method) # "GET"# ============ 其他 ============print(response.encoding) # 编码print(response.elapsed) # 耗时（timedelta对象）print(response.history) # 重定向历史print(response.cookies) # 响应中的Cookie
4.2 状态码速查
| | |
|---|
| 200 | | |
| 201 | | |
| 204 | | |
| 301 | | |
| 302 | | |
| 304 | | |
| 400 | | |
| 401 | | |
| 403 | | |
| 404 | | |
| 405 | | |
| 429 | | |
| 500 | | |
| 502 | | |
| 503 | | |
import requestsresponse = requests.get("https://httpbin.org/status/404")print(response.status_code) # 404print(response.ok) # False# 如果状态码表示错误，可以抛出异常response.raise_for_status()# 会抛出：requests.exceptions.HTTPError: 404 Client Error: NOT FOUND
4.3 text vs content vs json()
import requestsresponse = requests.get("https://httpbin.org/get")# text：字符串（适合 HTML、纯文本）print(type(response.text)) # <class 'str'>print(response.text[:100])# content：字节（适合图片、PDF、二进制文件）print(type(response.content)) # <class 'bytes'>print(response.content[:100])# json()：Python 字典/列表（适合 API 返回的 JSON）print(type(response.json())) # <class 'dict'>data = response.json()print(data["url"]) # "https://httpbin.org/get"# ⚠️ 如果响应不是 JSON，json() 会报错：# response = requests.get("https://www.baidu.com")# response.json() # ❌ JSONDecodeError
五、GET 请求详解
5.1 带查询参数（Query Parameters）
import requests# ============ 方法1：参数写在 URL 里 ============response = requests.get("https://httpbin.org/get?name=张三&age=25&city=北京")# ============ 方法2：用 params 参数（推荐！） ============params = { "name": "张三", "age": 25, "city": "北京", "hobbies": ["读书", "编程"], # 列表会变成 hobbies=读书&hobbies=编程}response = requests.get("https://httpbin.org/get", params=params)# requests 会自动：# 1. 把参数拼接到 URL 后面# 2. 进行 URL 编码（中文 → %E5%BC%A0%E4%B8%89）# 3. 处理特殊字符print(response.url)# https://httpbin.org/get?name=%E5%BC%A0%E4%B8%89&age=25&city=%E5%8C%97%E4%BA%AC&hobbies=%E8%AF%BB%E4%B9%A6&hobbies=%E7%BC%96%E7%A8%8B# 查看服务器收到的参数data = response.json()print(data["args"])# {'name': '张三', 'age': '25', 'city': '北京', 'hobbies': ['读书', '编程']}
5.2 实际例子：搜索
import requests# 模拟百度搜索params = { "wd": "Python requests 教程", # 搜索关键词 "ie": "utf-8", # 编码}response = requests.get("https://www.baidu.com/s", params=params)print(f"状态码：{response.status_code}")print(f"URL：{response.url}")# https://www.baidu.com/s?wd=Python+requests+%E6%95%99%E7%A8%8B&ie=utf-8# 检查是否包含搜索结果if "搜索结果" in response.text or "百度为您找到" in response.text: print("搜索成功！")
5.3 带请求头
import requests# 有些网站会检查 User-Agent，不带的话会返回 403headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive",}response = requests.get("https://www.example.com", headers=headers)print(response.status_code) # 200# 查看服务器收到的请求头response = requests.get("https://httpbin.org/headers", headers=headers)print(response.json()["headers"])
5.4 带 Cookie
import requests# 方法1：用 headers 传 Cookieheaders = { "Cookie": "session_id=abc123; user=zhangsan"}response = requests.get("https://httpbin.org/cookies", headers=headers)# 方法2：用 cookies 参数（推荐）cookies = { "session_id": "abc123", "user": "zhangsan",}response = requests.get("https://httpbin.org/cookies", cookies=cookies)print(response.json())# {"cookies": {"session_id": "abc123", "user": "zhangsan"}}
六、POST 请求详解
6.1 发送表单数据
import requests# POST 请求通常用于提交数据（登录、注册、提交表单）# ============ 发送表单数据（application/x-www-form-urlencoded） ============data = { "username": "zhangsan", "password": "123456", "remember": "true",}response = requests.post("https://httpbin.org/post", data=data)result = response.json()print(result["form"])# {'username': 'zhangsan', 'password': '123456', 'remember': 'true'}print(result["headers"]["Content-Type"])# application/x-www-form-urlencoded
6.2 发送 JSON 数据
import requests# ============ 发送 JSON（application/json） ============# 大多数现代 API 使用 JSON 格式payload = { "name": "张三", "age": 25, "skills": ["Python", "Flask", "SQL"], "address": { "city": "北京", "district": "海淀区", }}# 方法1：用 json 参数（推荐！自动序列化 + 设置Content-Type）response = requests.post("https://httpbin.org/post", json=payload)# 方法2：手动序列化（不推荐，除非有特殊需求）import jsonresponse = requests.post( "https://httpbin.org/post", data=json.dumps(payload), headers={"Content-Type": "application/json"})result = response.json()print(result["json"])# {'name': '张三', 'age': 25, 'skills': ['Python', 'Flask', 'SQL'], ...}print(result["headers"]["Content-Type"])# application/json
6.3 data vs json 参数对比
import requestsdata_dict = {"name": "张三", "age": 25}# ============ data 参数 ============# Content-Type: application/x-www-form-urlencoded# 数据格式：name=%E5%BC%A0%E4%B8%89&age=25response = requests.post("https://httpbin.org/post", data=data_dict)print(response.json()["form"]) # {'name': '张三', 'age': '25'}（注意age变成字符串了）print(response.json()["json"]) # None# ============ json 参数 ============# Content-Type: application/json# 数据格式：{"name": "张三", "age": 25}response = requests.post("https://httpbin.org/post", json=data_dict)print(response.json()["form"]) # {}print(response.json()["json"]) # {'name': '张三', 'age': 25}（age保持整数）# 总结：# 传统表单 → data# REST API → json
6.4 实际例子：调用 API
import requests# 模拟调用一个用户注册 APIapi_url = "https://httpbin.org/post"user_data = { "username": "newuser", "email": "newuser@example.com", "password": "SecurePass123!", "age": 28,}response = requests.post(api_url, json=user_data)if response.status_code == 200: result = response.json() print("注册成功！") print(f"服务器收到：{result['json']}")elif response.status_code == 400: print("参数错误！")elif response.status_code == 409: print("用户名已存在！")else: print(f"错误：{response.status_code}")
七、其他 HTTP 方法
import requestsurl = "https://httpbin.org"# ============ PUT：更新资源（全量更新） ============response = requests.put(f"{url}/put", json={"name": "李四", "age": 30})print(f"PUT: {response.status_code}")# ============ PATCH：更新资源（部分更新） ============response = requests.patch(f"{url}/patch", json={"age": 31})print(f"PATCH: {response.status_code}")# ============ DELETE：删除资源 ============response = requests.delete(f"{url}/delete", params={"id": 123})print(f"DELETE: {response.status_code}")# ============ HEAD：只获取响应头（不获取body） ============response = requests.head("https://www.baidu.com")print(f"HEAD: {response.status_code}")print(f"Content-Type: {response.headers.get('Content-Type')}")print(f"Body: '{response.text}'") # 空字符串（HEAD不返回body）# ============ OPTIONS：查看服务器支持的方法 ============response = requests.options("https://httpbin.org/get")print(f"Allow: {response.headers.get('Allow')}")# GET, HEAD, OPTIONS# ============ 通用方法 ============response = requests.request("GET", "https://httpbin.org/get")# 等价于 requests.get(...)
八、请求头（Headers）详解
8.1 常用请求头
import requestsheaders = { # 告诉服务器"我是谁"（模拟浏览器） "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", # 告诉服务器"我能接受什么格式的响应" "Accept": "application/json", # 告诉服务器"我发送的数据格式" "Content-Type": "application/json", # 认证令牌 "Authorization": "Bearer your_token_here", # 自定义头 "X-API-Key": "your-api-key", "X-Request-ID": "unique-id-123",}response = requests.get("https://httpbin.org/headers", headers=headers)print(response.json())
8.2 为什么需要 User-Agent？
import requests# 很多网站会检查 User-Agent# 如果不带，服务器可能认为你是爬虫，返回 403# ❌ 不带 User-Agent（可能被拒绝）response = requests.get("https://www.example.com")# requests 默认会带 "python-requests/2.32.3"# ✅ 模拟浏览器headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"}response = requests.get("https://www.example.com", headers=headers)print(response.status_code) # 200
九、Cookie 和 Session
9.1 Cookie 是什么？
Cookie = 服务器给浏览器的一张小纸条 浏览器下次请求时把纸条带回去 服务器通过纸条认出你场景： 1. 你登录网站 → 服务器发一个 session_id Cookie 2. 你浏览其他页面 → 浏览器自动带上 session_id 3. 服务器看到 session_id → "哦，这是张三"
9.2 手动管理 Cookie
import requests# 第一次请求：获取 Cookieresponse = requests.get("https://httpbin.org/cookies/set/session_id/abc123")print(f"收到的Cookie：{response.cookies}")# <RequestsCookieJar[<Cookie session_id=abc123 for httpbin.org/>]># 第二次请求：带上 Cookieresponse = requests.get( "https://httpbin.org/cookies", cookies=response.cookies # 把上次收到的 Cookie 传回去)print(response.json())# {"cookies": {"session_id": "abc123"}}
9.3 Session（自动管理 Cookie）⭐
import requests# Session 会自动保存和发送 Cookie（就像浏览器一样）session = requests.Session()# 第一次请求：服务器设置 Cookiesession.get("https://httpbin.org/cookies/set/user/zhangsan")session.get("https://httpbin.org/cookies/set/token/xyz789")# 后续请求：自动带上所有 Cookie（不需要手动传！）response = session.get("https://httpbin.org/cookies")print(response.json())# {"cookies": {"user": "zhangsan", "token": "xyz789"}}# Session 还可以设置默认 headers（所有请求都会带）session.headers.update({ "User-Agent": "MyApp/1.0", "Authorization": "Bearer my_token",})# 之后的所有请求都自动带这些 headersresponse = session.get("https://httpbin.org/headers")print(response.json()["headers"]["User-Agent"]) # "MyApp/1.0"# 用完关闭session.close()# 或者用 with 语句（推荐）with requests.Session() as session: session.get("https://httpbin.org/cookies/set/a/1") response = session.get("https://httpbin.org/cookies") print(response.json())
9.4 模拟登录（完整例子）
import requests# 模拟一个登录流程with requests.Session() as session: # 设置通用请求头 session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0", }) # 第1步：访问登录页面（获取 CSRF token 等） login_page = session.get("https://httpbin.org/get") # 实际场景中，这里会从页面中提取 csrf_token # 第2步：提交登录表单 login_data = { "username": "zhangsan", "password": "mypassword", # "csrf_token": csrf_token, # 实际场景需要 } login_response = session.post("https://httpbin.org/post", data=login_data) # 第3步：检查是否登录成功 if login_response.status_code == 200: print("登录成功！") # 第4步：访问需要登录的页面（Cookie 自动带上） profile = session.get("https://httpbin.org/cookies") print(f"当前Cookie：{profile.json()}")
十、认证（Authentication）
10.1 Basic Auth（基本认证）
import requestsfrom requests.auth import HTTPBasicAuth# 方法1：用 auth 参数response = requests.get( "https://httpbin.org/basic-auth/user/pass", auth=("user", "pass") # (用户名, 密码))print(response.status_code) # 200print(response.json()) # {"authenticated": true, "user": "user"}# 方法2：用 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", "wrong_password"))print(response.status_code) # 401
10.2 Bearer Token（最常用）
import requests# 大多数现代 API 使用 Bearer Token 认证token = "your_api_token_here"headers = { "Authorization": f"Bearer {token}",}response = requests.get("https://httpbin.org/bearer", headers=headers)print(response.status_code) # 200print(response.json()) # {"authenticated": true, "token": "your_api_token_here"}
10.3 API Key
import requests# 有些 API 用 API Key 认证# 方式1：放在 URL 参数中response = requests.get( "https://api.example.com/data", params={"api_key": "your_api_key"})# 方式2：放在请求头中（更常见）headers = { "X-API-Key": "your_api_key",}response = requests.get("https://api.example.com/data", headers=headers)
10.4 Digest Auth
import requestsfrom requests.auth import HTTPDigestAuthresponse = requests.get( "https://httpbin.org/digest-auth/auth/user/pass", auth=HTTPDigestAuth("user", "pass"))print(response.status_code) # 200
十一、文件上传和下载
11.1 上传文件
import requests# ============ 上传单个文件 ============# files 参数：{"字段名": ("文件名", 文件对象, "MIME类型")}with open("photo.jpg", "rb") as f: # ⚠️ 必须用 "rb"（二进制模式） files = { "file": ("photo.jpg", f, "image/jpeg"), } response = requests.post("https://httpbin.org/post", files=files)print(response.json()["files"])# {"file": "data:image/jpeg;base64,/9j/4AAQ...（base64编码的文件内容）"}# ============ 上传多个文件 ============files = [ ("files", ("file1.txt", open("file1.txt", "rb"), "text/plain")), ("files", ("file2.txt", open("file2.txt", "rb"), "text/plain")),]response = requests.post("https://httpbin.org/post", files=files)# ============ 上传文件 + 表单数据 ============with open("photo.jpg", "rb") as f: files = {"photo": ("photo.jpg", f, "image/jpeg")} data = { "title": "我的照片", "description": "一张风景照", "tags": "风景,旅行", } response = requests.post("https://httpbin.org/post", files=files, data=data)result = response.json()print(result["form"]) # {'title': '我的照片', 'description': '一张风景照', ...}print(result["files"]) # {'photo': 'data:image/jpeg;base64,...'}# ============ 上传字符串作为文件 ============files = { "file": ("data.csv", "name,age\n张三,25\n李四,30", "text/csv"),}response = requests.post("https://httpbin.org/post", files=files)
11.2 下载文件
import requests# ============ 下载小文件 ============response = requests.get("https://httpbin.org/image/png")# 保存到文件with open("downloaded_image.png", "wb") as f: # ⚠️ "wb" 二进制写入 f.write(response.content) # content 是 bytesprint(f"下载完成，大小：{len(response.content)} 字节")# ============ 下载大文件（流式下载） ============# 大文件不能一次性读入内存！用 stream=Trueurl = "https://example.com/large_file.zip"response = requests.get(url, stream=True) # ⚠️ stream=True# 检查是否成功response.raise_for_status()# 获取文件大小（如果服务器提供）total_size = int(response.headers.get("content-length", 0))print(f"文件大小：{total_size / 1024 / 1024:.2f} MB")# 分块写入downloaded = 0with open("large_file.zip", "wb") as f: for chunk in response.iter_content(chunk_size=8192): # 每次读 8KB if chunk: # 过滤 keep-alive 的空块 f.write(chunk) downloaded += len(chunk) # 显示进度 if total_size > 0: percent = downloaded / total_size * 100 print(f"\r下载进度：{percent:.1f}%", end="", flush=True)print("\n下载完成！")# ============ 通用下载函数 ============def download_file(url, filename, chunk_size=8192): """下载文件（带进度显示）""" response = requests.get(url, stream=True) response.raise_for_status() total = int(response.headers.get("content-length", 0)) downloaded = 0 with open(filename, "wb") as f: for chunk in response.iter_content(chunk_size=chunk_size): if chunk: f.write(chunk) downloaded += len(chunk) if total > 0: bar = "█" * int(downloaded / total * 40) print(f"\r|{bar:<40}| {downloaded/total*100:.1f}%", end="") print(f"\n✅ 已保存：{filename}（{downloaded} 字节）")# 使用# download_file("https://example.com/file.pdf", "output.pdf")
十二、超时和重试
12.1 设置超时
import requests# ============ 基本超时 ============# timeout 参数：等待服务器响应的最大秒数try: # 5秒内必须收到响应，否则报错 response = requests.get("https://httpbin.org/delay/1", timeout=5) print("成功！")except requests.exceptions.Timeout: print("超时了！服务器响应太慢")# ============ 分别设置连接超时和读取超时 ============# timeout=(连接超时, 读取超时)try: response = requests.get( "https://httpbin.org/get", timeout=(3, 10) # 连接最多等3秒，读取最多等10秒 )except requests.exceptions.ConnectTimeout: print("连接超时！服务器可能挂了")except requests.exceptions.ReadTimeout: print("读取超时！服务器处理太慢")# ============ 测试超时 ============try: # httpbin.org/delay/10 会等10秒才响应 response = requests.get("https://httpbin.org/delay/10", timeout=2)except requests.exceptions.Timeout: print("果然超时了（2秒限制，服务器要10秒）")
12.2 重试机制
import requestsfrom requests.adapters import HTTPAdapterfrom urllib3.util.retry import Retry# ============ 方法1：简单重试（手动） ============def fetch_with_retry(url, max_retries=3, timeout=10): """带重试的请求""" for attempt in range(1, max_retries + 1): try: response = requests.get(url, timeout=timeout) response.raise_for_status() return response except requests.exceptions.RequestException as e: print(f"第 {attempt} 次尝试失败：{e}") if attempt == max_retries: raise # 最后一次还失败，抛出异常 import time time.sleep(2 ** attempt) # 指数退避：2s, 4s, 8s...# ============ 方法2：用 HTTPAdapter（推荐） ============session = requests.Session()# 配置重试策略retry_strategy = Retry( total=3, # 最多重试3次 backoff_factor=1, # 退避因子：1s, 2s, 4s status_forcelist=[500, 502, 503, 504], # 这些状态码触发重试 allowed_methods=["GET", "POST"], # 这些方法允许重试)# 挂载到 sessionadapter = HTTPAdapter(max_retries=retry_strategy)session.mount("http://", adapter)session.mount("https://", adapter)# 使用response = session.get("https://httpbin.org/get", timeout=10)print(response.status_code)session.close()
十三、错误处理
13.1 完整的异常处理
import requestsdef safe_request(url, method="GET", **kwargs): """安全的请求函数（包含完整错误处理）""" # 设置默认超时 kwargs.setdefault("timeout", 10) try: # 发送请求 if method.upper() == "GET": response = requests.get(url, **kwargs) elif method.upper() == "POST": response = requests.post(url, **kwargs) else: response = requests.request(method, url, **kwargs) # 检查 HTTP 状态码（4xx, 5xx 会抛异常） response.raise_for_status() return response except requests.exceptions.Timeout: print(f"❌ 超时：{url}") return None except requests.exceptions.ConnectionError: print(f"❌ 连接错误：无法连接到 {url}") return None except requests.exceptions.HTTPError as e: print(f"❌ HTTP错误：{e.response.status_code} - {e.response.reason}") return None except requests.exceptions.TooManyRedirects: print(f"❌ 重定向次数过多：{url}") return None except requests.exceptions.RequestException as e: print(f"❌ 请求异常：{e}") return None# 使用response = safe_request("https://httpbin.org/get")if response: print(f"✅ 成功：{response.status_code}")response = safe_request("https://httpbin.org/status/404")# ❌ HTTP错误：404 - NOT FOUNDresponse = safe_request("https://this-domain-does-not-exist-12345.com")# ❌ 连接错误：无法连接到 ...
13.2 异常层级
requests.exceptions.RequestException（基类）├── ConnectionError（连接错误）│ └── ConnectTimeout（连接超时）├── Timeout（超时）│ ├── ConnectTimeout（连接超时）│ └── ReadTimeout（读取超时）├── HTTPError（HTTP状态码错误，raise_for_status()抛出）├── TooManyRedirects（重定向过多）├── URLRequired（URL缺失）└── MissingSchema（URL格式错误）
十四、JSON 处理
14.1 解析 JSON 响应
import requests# 调用一个返回 JSON 的 APIresponse = requests.get("https://httpbin.org/get")# 方法1：用 response.json()（推荐）data = response.json()print(type(data)) # <class 'dict'>print(data["url"]) # "https://httpbin.org/get"print(data["headers"]) # 请求头字典# 方法2：手动解析import jsondata = json.loads(response.text)# ⚠️ 如果响应不是 JSON，json() 会报错：try: data = response.json()except json.JSONDecodeError: print("响应不是有效的JSON") print(response.text[:200]) # 看看实际返回了什么
14.2 发送 JSON 请求
import requests# 调用一个需要 JSON body 的 APIapi_url = "https://httpbin.org/post"payload = { "model": "gpt-4", "messages": [ {"role": "system", "content": "你是一个助手"}, {"role": "user", "content": "你好！"}, ], "temperature": 0.7, "max_tokens": 100,}response = requests.post(api_url, json=payload)# json 参数会自动：# 1. 把 dict 序列化为 JSON 字符串# 2. 设置 Content-Type: application/jsonprint(response.json()["json"]) # 服务器收到的数据
14.3 处理嵌套 JSON
import requests# 假设 API 返回复杂嵌套数据response = requests.get("https://httpbin.org/get")data = response.json()# 安全地访问嵌套字段# ❌ 危险写法（如果某层不存在会报 KeyError）# city = data["address"]["city"]# ✅ 安全写法city = data.get("address", {}).get("city", "未知")# 或者用 try/excepttry: city = data["address"]["city"]except (KeyError, TypeError): city = "未知"
十五、重定向
import requests# ============ 默认行为：自动跟随重定向 ============response = requests.get("https://httpbin.org/redirect-to?url=https://www.baidu.com")print(response.status_code) # 200（最终页面的状态码）print(response.url) # https://www.baidu.com（最终URL）print(response.history) # 重定向历史# 查看重定向链for r in response.history: print(f" {r.status_code} → {r.url}")# 302 → https://httpbin.org/redirect-to?url=...# ============ 禁止自动重定向 ============response = requests.get( "https://httpbin.org/redirect-to?url=https://www.baidu.com", allow_redirects=False # 不跟随重定向)print(response.status_code) # 302（重定向状态码）print(response.headers["Location"]) # https://www.baidu.com（目标地址）# ============ 限制重定向次数 ============try: response = requests.get( "https://httpbin.org/redirect/10", # 重定向10次 allow_redirects=True, # requests 默认最多重定向 30 次 )except requests.exceptions.TooManyRedirects: print("重定向次数过多！")
十六、代理（Proxy）
import requests# ============ 设置代理 ============proxies = { "http": "http://127.0.0.1:7890", "https": "http://127.0.0.1:7890",}response = requests.get("https://httpbin.org/ip", proxies=proxies)print(response.json()) # 显示代理服务器的IP# ============ 带认证的代理 ============proxies = { "http": "http://user:password@proxy.example.com:8080", "https": "http://user:password@proxy.example.com:8080",}# ============ SOCKS 代理 ============# 需要安装：pip install requests[socks]proxies = { "http": "socks5://127.0.0.1:1080", "https": "socks5://127.0.0.1:1080",}# ============ 在 Session 中设置（所有请求都用代理） ============session = requests.Session()session.proxies = { "http": "http://127.0.0.1:7890", "https": "http://127.0.0.1:7890",}
十七、SSL 证书
import requests# ============ 默认：验证 SSL 证书 ============response = requests.get("https://www.baidu.com") # 正常# ============ 跳过 SSL 验证（不推荐！仅测试用） ============response = requests.get("https://self-signed.example.com", verify=False)# 会有警告：InsecureRequestWarning# 消除警告：import urllib3urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)# ============ 使用自定义 CA 证书 ============response = requests.get("https://example.com", verify="/path/to/ca-bundle.crt")# ============ 客户端证书（双向认证） ============response = requests.get( "https://example.com", cert=("/path/to/client.crt", "/path/to/client.key"))
十八、高级用法
18.1 自定义请求（Prepared Request）
import requests# 当你需要更精细的控制时session = requests.Session()# 创建请求对象req = requests.Request( method="POST", url="https://httpbin.org/post", json={"key": "value"}, headers={"X-Custom": "header"},)# 准备请求（编码参数、设置headers等）prepared = session.prepare_request(req)# 可以修改 prepared 对象prepared.headers["X-Extra"] = "added_later"# 发送response = session.send(prepared, timeout=10)print(response.json())
18.2 事件钩子（Hooks）
import requests# 钩子：在请求/响应的特定时刻执行自定义代码def print_url(response, *args, **kwargs): """响应到达时打印URL""" print(f"收到响应：{response.url} [{response.status_code}]")def print_elapsed(response, *args, **kwargs): """打印耗时""" print(f"耗时：{response.elapsed.total_seconds():.3f}s")# 使用钩子response = requests.get( "https://httpbin.org/get", hooks={"response": [print_url, print_elapsed]})# 输出：# 收到响应：https://httpbin.org/get [200]# 耗时：0.234s
18.3 自定义适配器
import requestsfrom requests.adapters import HTTPAdapter# 自定义连接池大小session = requests.Session()adapter = HTTPAdapter( pool_connections=10, # 连接池数量 pool_maxsize=20, # 每个连接池最大连接数 max_retries=3, # 重试次数)session.mount("http://", adapter)session.mount("https://", adapter)# 使用response = session.get("https://httpbin.org/get")
18.4 流式请求（SSE / 大文件）
import requests# ============ 流式读取（Server-Sent Events） ============response = requests.get("https://httpbin.org/stream/5", stream=True)for line in response.iter_lines(): if line: # 过滤空行 print(line.decode("utf-8"))# ============ 流式读取（逐行） ============response = requests.get("https://example.com/large.txt", stream=True)for chunk in response.iter_content(chunk_size=1024): process(chunk) # 处理每一块# ============ 流式上传 ============# 上传大文件时，可以流式读取def file_stream(filename, chunk_size=8192): with open(filename, "rb") as f: while chunk := f.read(chunk_size): yield chunk# requests 会自动处理生成器with open("large_file.bin", "rb") as f: response = requests.post("https://httpbin.org/post", data=f)
十九、实战案例
19.1 天气查询
import requestsdef get_weather(city): """查询城市天气（使用 wttr.in 免费API）""" url = f"https://wttr.in/{city}" params = {"format": "j1"} # JSON 格式 headers = {"User-Agent": "curl/7.68.0"} try: response = requests.get(url, params=params, headers=headers, timeout=10) response.raise_for_status() data = response.json() current = data["current_condition"][0] area = data["nearest_area"][0] print(f"📍 {area['areaName'][0]['value']}, {area['country'][0]['value']}") print(f"🌡️ 温度：{current['temp_C']}°C（体感 {current['FeelsLikeC']}°C）") print(f"💧 湿度：{current['humidity']}%") print(f"🌬️ 风速：{current['windspeedKmph']} km/h") print(f"☁️ 天气：{current['weatherDesc'][0]['value']}") except requests.exceptions.RequestException as e: print(f"查询失败：{e}")get_weather("Beijing")
19.2 GitHub API
import requestsdef get_github_user(username): """获取 GitHub 用户信息""" url = f"https://api.github.com/users/{username}" headers = { "Accept": "application/vnd.github.v3+json", # 如果有 token： # "Authorization": "Bearer YOUR_TOKEN", } response = requests.get(url, headers=headers, timeout=10) if response.status_code == 200: user = response.json() print(f"👤 {user['name'] or user['login']}") print(f"📝 Bio: {user.get('bio', 'N/A')}") print(f"📍 位置: {user.get('location', 'N/A')}") print(f"👥 关注者: {user['followers']}") print(f"📦 公开仓库: {user['public_repos']}") print(f"🔗 {user['html_url']}") elif response.status_code == 404: print(f"用户 {username} 不存在") elif response.status_code == 403: print("API 限流了，请稍后再试") else: print(f"错误：{response.status_code}")def get_github_repos(username, sort="stars", per_page=5): """获取用户的热门仓库""" url = f"https://api.github.com/users/{username}/repos" params = { "sort": sort, "direction": "desc", "per_page": per_page, } response = requests.get(url, params=params, timeout=10) if response.status_code == 200: repos = response.json() print(f"\n📦 {username} 的热门仓库：") for repo in repos: stars = repo["stargazers_count"] lang = repo.get("language") or "N/A" print(f" ⭐ {stars:>6} | {repo['name']:<30} | {lang}")get_github_user("torvalds")get_github_repos("torvalds")
19.3 简单的网页爬虫
import requestsimport redef scrape_quotes(): """爬取名言（httpbin.org 示例）""" url = "https://httpbin.org/html" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0", } response = requests.get(url, headers=headers, timeout=10) response.raise_for_status() # 简单的正则提取（实际项目建议用 BeautifulSoup） html = response.text # 提取标题 title_match = re.search(r"<h1>(.*?)</h1>", html) if title_match: print(f"标题：{title_match.group(1)}") # 提取段落 paragraphs = re.findall(r"<p>(.*?)</p>", html, re.DOTALL) for i, p in enumerate(paragraphs[:3], 1): # 去掉HTML标签 clean = re.sub(r"<[^>]+>", "", p).strip() print(f"段落{i}：{clean[:100]}...")scrape_quotes()
19.4 批量请求（并发）
import requestsimport concurrent.futuresimport timedef fetch_url(url): """获取单个URL""" try: response = requests.get(url, timeout=10) return url, response.status_code, len(response.text) except requests.exceptions.RequestException as e: return url, str(e), 0# 要请求的URL列表urls = [ "https://httpbin.org/get", "https://httpbin.org/delay/1", "https://httpbin.org/status/200", "https://httpbin.org/status/404", "https://httpbin.org/ip",]# ============ 串行（慢） ============start = time.time()for url in urls: result = fetch_url(url) print(f" {result[1]:>4} | {result[0]}")print(f"串行耗时：{time.time() - start:.2f}s")# ============ 并发（快） ============start = time.time()with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: futures = {executor.submit(fetch_url, url): url for url in urls} for future in concurrent.futures.as_completed(futures): url, status, size = future.result() print(f" {status:>4} | {url} ({size} bytes)")print(f"并发耗时：{time.time() - start:.2f}s")
19.5 REST API 客户端封装
import requestsimport jsonclass APIClient: """通用 REST API 客户端""" def __init__(self, base_url, api_key=None, timeout=30): self.base_url = base_url.rstrip("/") self.timeout = timeout self.session = requests.Session() # 设置默认请求头 self.session.headers.update({ "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "MyAPIClient/1.0", }) # 如果有 API Key if api_key: self.session.headers["Authorization"] = f"Bearer {api_key}" def _request(self, method, endpoint, **kwargs): """发送请求的通用方法""" url = f"{self.base_url}/{endpoint.lstrip('/')}" kwargs.setdefault("timeout", self.timeout) try: response = self.session.request(method, url, **kwargs) response.raise_for_status() # 尝试解析 JSON if response.content: return response.json() return None except requests.exceptions.HTTPError as e: error_body = e.response.text raise Exception(f"API错误 [{e.response.status_code}]: {error_body}") except requests.exceptions.Timeout: raise Exception(f"请求超时：{url}") except requests.exceptions.ConnectionError: raise Exception(f"无法连接到服务器：{url}") def get(self, endpoint, params=None): return self._request("GET", endpoint, params=params) def post(self, endpoint, data=None): return self._request("POST", endpoint, json=data) def put(self, endpoint, data=None): return self._request("PUT", endpoint, json=data) def patch(self, endpoint, data=None): return self._request("PATCH", endpoint, json=data) def delete(self, endpoint): return self._request("DELETE", endpoint) def close(self): self.session.close() def __enter__(self): return self def __exit__(self, *args): self.close()# ============ 使用示例 ============with APIClient("https://httpbin.org") as api: # GET result = api.get("/get", params={"name": "test"}) print(f"GET: {result['args']}") # POST result = api.post("/post", data={"username": "zhangsan", "age": 25}) print(f"POST: {result['json']}") # PUT result = api.put("/put", data={"name": "李四"}) print(f"PUT: {result['json']}")
二十、最佳实践
20.1 代码规范
import requests# ✅ 好的写法def fetch_data(url, params=None, timeout=10): """获取数据（带完整错误处理）""" try: response = requests.get(url, params=params, timeout=timeout) response.raise_for_status() return response.json() except requests.exceptions.Timeout: print(f"超时：{url}") return None except requests.exceptions.HTTPError as e: print(f"HTTP错误：{e.response.status_code}") return None except requests.exceptions.RequestException as e: print(f"请求失败：{e}") return None# ❌ 不好的写法def bad_fetch(url): r = requests.get(url) # 没有超时！没有错误处理！ return r.json() # 如果不是JSON会崩溃！
20.2 注意事项清单
# 1. ✅ 永远设置 timeoutrequests.get(url, timeout=10)# 2. ✅ 检查状态码response.raise_for_status()# 或if response.status_code == 200: ...# 3. ✅ 用 Session 复用连接（多次请求同一服务器）with requests.Session() as s: s.get(url1) s.get(url2) s.get(url3)# 4. ✅ 大文件用 stream=Trueresponse = requests.get(url, stream=True)for chunk in response.iter_content(8192): ...# 5. ✅ 用 json 参数发 JSON（不要手动 dumps）requests.post(url, json=data) # ✅# requests.post(url, data=json.dumps(data)) # ❌ 还要手动设header# 6. ✅ 文件用 "rb" 模式打开withopen("file.jpg", "rb") as f: requests.post(url, files={"file": f})# 7. ❌ 不要在生产环境 verify=False# requests.get(url, verify=False) # 不安全！# 8. ✅ 礼貌爬取（加延迟）import timefor url in urls: requests.get(url) time.sleep(1) # 每次请求间隔1秒
20.3 性能优化
import requests# 1. 使用 Session（TCP连接复用，减少握手开销）session = requests.Session()for i in range(100): session.get("https://example.com/api/data")session.close()# 2. 禁用不需要的功能session = requests.Session()session.trust_env = False # 不读取环境变量中的代理设置# 3. 连接池调优from requests.adapters import HTTPAdapteradapter = HTTPAdapter(pool_connections=10, pool_maxsize=20)session.mount("https://", adapter)# 4. 只获取需要的数据# 如果只需要 headers，用 HEAD 请求response = requests.head(url) # 不下载 body# 5. 使用 gzip 压缩（requests 默认启用）# 服务器返回压缩数据，requests 自动解压
二十一、requests 方法参数速查表
requests.get(url, params=None, headers=None, cookies=None, auth=None, timeout=None, allow_redirects=True, proxies=None, verify=True, stream=False, cert=None, json=None)requests.post(url, data=None, json=None, headers=None, cookies=None, files=None, auth=None, timeout=None, allow_redirects=True, proxies=None, verify=True, stream=False)
| | |
|---|
url | | |
params | | |
data | | |
json | | |
headers | | |
cookies | | |
files | | |
auth | | |
timeout | | |
allow_redirects | | |
proxies | | |
verify | | |
stream | | |
cert | | |
hooks | | |
二十二、Response 属性速查表
| | |
|---|
response.status_code | | |
response.ok | | |
response.text | | |
response.content | | |
response.json() | | |
response.headers | | |
response.url | | |
response.encoding | | |
response.elapsed | | |
response.history | | |
response.cookies | | |
response.request | | |
response.raise_for_status() | | |
response.iter_content() | | |
response.iter_lines() | | |
二十三、学习路径建议
第1天：安装 + 第一个 GET 请求 + Response 对象第2天：params 参数 + headers + 状态码处理第3天：POST 请求（data vs json）第4天：Session + Cookie（模拟登录）第5天：文件上传/下载第6天：错误处理 + 超时 + 重试第7天：实战（调API、简单爬虫）第8天：进阶（并发、代理、流式）
二十四、一句话总结
requests = 用 Python 和互联网对话的最简方式
记住三个核心：
requests.get(url) / requests.post(url, json=data) —— 发请求response.status_code / response.json() / response.text —— 读响应- 永远设置
timeout，永远处理异常
