Python Requests Session 对象详解

什么是Session对象？

在Requests库中，Session对象允许你在多个HTTP请求之间保持某些参数，它会自动处理Cookie，并在同一个Session实例发出的所有请求之间保持连接，从而重用底层的TCP连接。

import requests

# 创建Session对象
session = requests.Session()

# 设置默认参数
session.headers.update({'User-Agent': 'MyApp/1.0'})

# 发送多个请求，它们会共享相同的Session设置
response1 = session.get('https://httpbin.org/cookies/set/sessioncookie/123456789')
response2 = session.get('https://httpbin.org/cookies')

print("第一个请求的Cookie:", response1.cookies)
print("第二个请求的Cookie:", response2.cookies)
print("第二个请求的响应内容:", response2.json())

为什么使用Session对象？

与直接使用requests.get()或requests.post()相比，Session对象提供了更好的性能和更方便的功能。

Session对象基本使用

1. 创建Session对象

import requests

# 创建Session对象
session = requests.Session()

# 或者使用上下文管理器（推荐，自动关闭连接）
with requests.Session() as session:
    response = session.get('https://httpbin.org/get')
    print(response.status_code)

2. 设置默认参数

可以为Session对象设置默认参数，这些参数会应用到该Session发出的所有请求：

import requests

session = requests.Session()

# 设置默认headers
session.headers.update({
    'User-Agent': 'MyApp/1.0',
    'Accept': 'application/json',
    'Accept-Language': 'zh-CN,zh;q=0.9',
    'Content-Type': 'application/json'
})

# 发送请求，会自动使用上面设置的headers
response = session.get('https://httpbin.org/headers')
print(response.json()['headers'])

import requests
from requests.auth import HTTPBasicAuth

session = requests.Session()

# 设置基本认证
session.auth = HTTPBasicAuth('username', 'password')

# 或者设置Bearer Token
# session.headers.update({'Authorization': 'Bearer YOUR_TOKEN'})

# 所有请求都会自动携带认证信息
response = session.get('https://httpbin.org/basic-auth/username/password')
print(response.status_code)  # 200

import requests

session = requests.Session()

# 设置代理
session.proxies = {
    'http': 'http://10.10.1.10:3128',
    'https': 'http://10.10.1.10:1080',
}

# 或者设置代理认证
# session.proxies = {'https': 'http://user:pass@10.10.1.10:3128/'}

response = session.get('https://httpbin.org/ip')
print(response.json())

3. Cookie管理示例

Session对象会自动处理Cookie，无需手动管理：

import requests

# 模拟登录流程
session = requests.Session()

# 1. 首先访问登录页面获取初始Cookie
login_page = session.get('https://example.com/login')
print("初始Cookie:", session.cookies.get_dict())

# 2. 提交登录表单
login_data = {
    'username': 'user123',
    'password': 'pass123'
}
login_response = session.post('https://example.com/login', data=login_data)
print("登录后Cookie:", session.cookies.get_dict())

# 3. 访问需要登录的页面（自动携带Cookie）
profile_response = session.get('https://example.com/profile')
print("访问个人页面状态码:", profile_response.status_code)

# 4. 手动操作Cookie
# 添加Cookie
session.cookies.set('custom_cookie', 'value123')

# 获取特定Cookie值
cookie_value = session.cookies.get('sessionid')
print("sessionid值:", cookie_value)

# 删除Cookie
session.cookies.clear('custom_cookie')

Session对象高级功能

1. 连接适配器配置

可以配置连接适配器来调整连接行为：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()

# 创建重试策略
retry_strategy = Retry(
    total=3,  # 总重试次数
    backoff_factor=1,  # 重试等待时间因子
    status_forcelist=[429, 500, 502, 503, 504],  # 遇到这些状态码重试
    allowed_methods=["GET", "POST"]  # 允许重试的HTTP方法
)

# 创建适配器并应用重试策略
adapter = HTTPAdapter(max_retries=retry_strategy)

# 为http和https连接挂载适配器
session.mount("http://", adapter)
session.mount("https://", adapter)

# 现在请求会自动重试
try:
    response = session.get("https://httpbin.org/status/500")
except requests.exceptions.RetryError as e:
    print("重试次数用完，请求失败:", e)

2. 超时设置

可以为Session设置全局超时，也可以为单个请求设置超时：

import requests

session = requests.Session()

# 方法1：为每个请求单独设置超时（推荐）
response = session.get('https://httpbin.org/delay/5', timeout=3.0)  # 3秒超时

# 方法2：设置Session级别的默认超时
# 这会影响该Session发出的所有请求
session.timeout = 5.0  # 5秒超时

# 超时可以分别设置连接超时和读取超时
try:
    response = session.get(
        'https://httpbin.org/delay/10',
        timeout=(3.05, 27)  # (连接超时, 读取超时)
    )
except requests.exceptions.Timeout:
    print("请求超时")

3. 事件钩子

可以使用事件钩子在请求的生命周期中执行自定义代码：

import requests

def print_url(r, *args, **kwargs):
    print(f"请求URL: {r.url}")

def check_status(r, *args, **kwargs):
    r.raise_for_status()  # 如果状态码不是200，抛出异常

session = requests.Session()

# 注册钩子函数
session.hooks['response'] = [print_url, check_status]

# 发送请求，钩子函数会自动执行
response = session.get('https://httpbin.org/status/404')
print(f"状态码: {response.status_code}")

4. Session对象的属性和方法

• session.headers - 默认的请求头字典
• session.cookies - CookieJar对象，存储所有Cookie
• session.auth - 默认的认证信息
• session.proxies - 代理设置字典
• session.params - 查询参数的默认字典
• session.stream - 是否使用流式响应（默认False）
• session.verify - SSL证书验证设置
• session.cert - 客户端SSL证书
• session.max_redirects - 最大重定向次数
• session.trust_env - 是否信任环境变量中的代理设置
• session.prepare_request(request) - 准备请求对象
• session.send(request, **kwargs) - 发送请求对象
• session.merge_environment_settings() - 合并环境设置

Session性能优化

import requests
import time

def test_without_session():
    """不使用Session，发送10个请求"""
    start = time.time()
    for i in range(10):
        requests.get('https://httpbin.org/get')
    return time.time() - start

def test_with_session():
    """使用Session，发送10个请求"""
    start = time.time()
    with requests.Session() as session:
        for i in range(10):
            session.get('https://httpbin.org/get')
    return time.time() - start

time_without = test_without_session()
time_with = test_with_session()

print(f"不使用Session: {time_without:.2f}秒")
print(f"使用Session: {time_with:.2f}秒")
print(f"性能提升: {(time_without - time_with) / time_without * 100:.1f}%")

优化建议

from requests.adapters import HTTPAdapter

session = requests.Session()

# 创建适配器并设置连接池大小
adapter = HTTPAdapter(pool_connections=10, pool_maxsize=100)

# 挂载适配器
session.mount('http://', adapter)
session.mount('https://', adapter)

# 不推荐：为每个请求创建新Session
for url in urls:
    response = requests.get(url)  # 每次都创建新连接

# 推荐：创建一次Session并重用
session = requests.Session()
for url in urls:
    response = session.get(url)  # 重用连接

实际应用示例

1. 网站爬虫示例

import requests
from bs4 import BeautifulSoup
import time

class WebsiteCrawler:
    def __init__(self):
        self.session = requests.Session()
        # 设置请求头，模拟浏览器
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
        })

    def login(self, username, password):
        """模拟登录"""
        login_url = 'https://example.com/login'

        # 首先获取登录页面，获取CSRF token等
        response = self.session.get(login_url)

        # 假设需要提交表单数据
        login_data = {
            'username': username,
            'password': password,
            # 可以从页面中提取其他必要字段
        }

        # 提交登录表单
        login_response = self.session.post(login_url, data=login_data)

        # 检查登录是否成功
        if '欢迎' in login_response.text:
            print("登录成功")
            return True
        else:
            print("登录失败")
            return False

    def crawl_pages(self, start_url, max_pages=10):
        """爬取多个页面"""
        pages_crawled = 0
        current_url = start_url

        while pages_crawled < max_pages and current_url:
            try:
                response = self.session.get(current_url, timeout=10)
                response.raise_for_status()

                # 解析页面内容
                soup = BeautifulSoup(response.text, 'html.parser')

                # 提取需要的数据
                self.extract_data(soup)

                # 寻找下一页链接
                next_link = soup.find('a', text='下一页')
                if next_link and 'href' in next_link.attrs:
                    current_url = next_link['href']
                else:
                    break

                pages_crawled += 1

                # 礼貌性延迟，避免对服务器造成压力
                time.sleep(1)

            except requests.exceptions.RequestException as e:
                print(f"请求失败: {e}")
                break

        print(f"共爬取 {pages_crawled} 个页面")

    def extract_data(self, soup):
        """提取页面数据（示例）"""
        # 这里添加具体的数据提取逻辑
        titles = soup.find_all('h2', class_='title')
        for title in titles:
            print(title.text.strip())

    def close(self):
        """关闭Session"""
        self.session.close()

# 使用示例
crawler = WebsiteCrawler()
if crawler.login('username', 'password'):
    crawler.crawl_pages('https://example.com/page1')
crawler.close()

2. API客户端示例

import requests
import json

class APIClient:
    def __init__(self, base_url, api_key=None):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()

        # 设置默认请求头
        headers = {
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        }

        if api_key:
            headers['Authorization'] = f'Bearer {api_key}'

        self.session.headers.update(headers)

        # 配置重试策略
        from requests.adapters import HTTPAdapter
        from urllib3.util.retry import Retry

        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[500, 502, 503, 504]
        )

        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

    def get(self, endpoint, params=None):
        """发送GET请求"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.get(url, params=params)
        response.raise_for_status()
        return response.json()

    def post(self, endpoint, data=None):
        """发送POST请求"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.post(url, json=data)
        response.raise_for_status()
        return response.json()

    def put(self, endpoint, data=None):
        """发送PUT请求"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.put(url, json=data)
        response.raise_for_status()
        return response.json()

    def delete(self, endpoint):
        """发送DELETE请求"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.delete(url)
        response.raise_for_status()
        return response.json()

    def close(self):
        """关闭Session"""
        self.session.close()

# 使用示例
# 使用上下文管理器自动管理资源
with APIClient('https://api.example.com', 'your_api_key') as client:
    # 获取用户列表
    users = client.get('/users')
    print(f"获取到 {len(users)} 个用户")

    # 创建新用户
    new_user = {'name': '张三', 'email': 'zhangsan@example.com'}
    created_user = client.post('/users', new_user)
    print(f"创建用户: {created_user['id']}")

    # 更新用户
    updated_data = {'email': 'newemail@example.com'}
    updated_user = client.put(f'/users/{created_user["id"]}', updated_data)
    print(f"更新用户: {updated_user['email']}")