Python网络请求库Requests详解：从基础认证到高级应用

2026年3月19日 互联网

一、HTTP基础认证：从原理到实践

HTTP基础认证（Basic Authentication）作为Web服务最常用的身份验证机制，其核心原理是通过Base64编码传输用户名密码。当服务端配置Basic Auth后，客户端需在请求头中添加Authorization字段，格式为Basic <credentials>，其中credentials是username:password的Base64编码结果。

1.1 基础认证实现

使用Requests库实现基础认证非常简单，通过auth参数即可完成：

  1. import requests
  2. from requests.auth import HTTPBasicAuth
  4. response = requests.get(
  5.     'https://api.example.com/protected',
  6.     auth=HTTPBasicAuth('username', 'password')
  7. )
  8. print(response.status_code)  # 200表示认证成功

更简洁的写法是直接传递元组：

  1. response = requests.get(
  2.     'https://api.example.com/protected',
  3.     auth=('username', 'password')
  4. )

1.2 安全注意事项

尽管实现简单，但需注意：

  • 传输安全：Basic Auth必须配合HTTPS使用，否则明文传输的凭证易被截获
  • 存储安全：避免在代码中硬编码凭证，建议使用环境变量或配置管理工具
  • 会话管理：频繁认证会降低性能，建议结合Session对象复用连接

二、会话管理与状态保持

Requests的Session对象能自动处理cookies和连接池，特别适合需要维持会话的场景：

2.1 会话对象基础

  1. with requests.Session() as session:
  2.     # 首次登录获取会话
  3.     login_response = session.post(
  4.         'https://api.example.com/login',
  5.         data={'username': 'user', 'password': 'pass'}
  6.     )
  8.     # 后续请求自动携带会话信息
  9.     data_response = session.get('https://api.example.com/data')
  10.     print(data_response.json())

2.2 高级配置技巧

  • 超时设置：通过timeout参数避免请求挂起 
    1. session.request('GET', url, timeout=(3.05, 27))  # 连接超时3.05秒，读取超时27秒
  • 适配器定制：控制连接池大小和重试策略
    ```python
    from requests.adapters import HTTPAdapter
    from urllib3.util.retry import Retry

retry_strategy = Retry(
total=3,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=[“HEAD”, “GET”, “OPTIONS”]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount(“https://“, adapter)

  2. # 三、异常处理与调试技巧
  3. ## 3.1 异常类型体系
  4. Requests定义了完善的异常层次：
  5. - `requests.exceptions.RequestException`：所有异常基类
  6. - `ConnectionError`：网络连接问题
  7. - `HTTPError`：HTTP错误状态码（4XX/5XX）
  8. - `Timeout`：请求超时
  9. - `TooManyRedirects`：重定向循环
  11. ## 3.2 健壮性处理示例
  12. ```python
  13. try:
  14.     response = requests.get('https://api.example.com/data', timeout=5)
  15.     response.raise_for_status()  # 自动触发HTTPError
  16.     data = response.json()
  17. except requests.exceptions.HTTPError as errh:
  18.     print(f"HTTP Error: {errh}")
  19. except requests.exceptions.ConnectionError as errc:
  20.     print(f"Connection Error: {errc}")
  21. except requests.exceptions.Timeout as errt:
  22.     print(f"Timeout Error: {errt}")
  23. except requests.exceptions.RequestException as err:
  24.     print(f"Unexpected Error: {err}")
  25. else:
  26.     print("Request succeeded")

3.3 调试工具推荐

  • 日志记录：启用Requests内置日志 
    1. import logging
    2. logging.basicConfig(level=logging.DEBUG)
  • 请求验证：使用requests.Request对象预构建请求 
    1. req = requests.Request('GET', 'https://api.example.com', 
    2.                     auth=('user', 'pass'),
    3.                     headers={'X-Custom': 'value'})
    4. prepared = req.prepare()
    5. print(prepared.headers)  # 查看最终请求头

四、企业级应用场景

4.1 API调用最佳实践

  • 批量请求优化：使用多线程/异步提高吞吐量
    ```python
    from concurrent.futures import ThreadPoolExecutor

urls = […] # 多个API端点
with ThreadPoolExecutor(max_workers=5) as executor:
results = list(executor.map(requests.get, urls))

  1. - **速率限制处理**：结合令牌桶算法控制请求频率
  2. ```python
  3. import time
  4. from collections import deque
  6. class RateLimiter:
  7.     def __init__(self, rate, per):
  8.         self.tokens = deque()
  9.         self.rate = rate
  10.         self.per = per
  11.         self.refresh()
  13.     def refresh(self):
  14.         now = time.time()
  15.         while self.tokens and self.tokens[0] <= now - self.per:
  16.             self.tokens.popleft()
  17.         while len(self.tokens) < self.rate:
  18.             self.tokens.append(time.time())
  20.     def wait(self):
  21.         self.refresh()
  22.         sleep_time = self.tokens[0] + self.per - time.time()
  23.         if sleep_time > 0:
  24.             time.sleep(sleep_time)
  25.         self.tokens.popleft()
  26.         self.tokens.append(time.time())
  28. limiter = RateLimiter(10, 1)  # 每秒10次
  29. for _ in range(100):
  30.     limiter.wait()
  31.     requests.get('https://api.example.com')

4.2 安全增强方案

  • 证书验证：严格校验服务端证书 
    1. response = requests.get('https://api.example.com', verify='/path/to/cert.pem')
  • 自定义证书池：针对自签名证书场景
    ```python
    import ssl
    from requests.adapters import HTTPAdapter

class CustomAdapter(HTTPAdapter):
def init_poolmanager(self, args, **kwargs):
context = ssl.create_default_context()
context.load_verify_locations(‘/path/to/custom_certs.pem’)
kwargs[‘ssl_context’] = context
super().init_poolmanager(args, **kwargs)

session = requests.Session()
session.mount(‘https://‘, CustomAdapter())

  2. # 五、性能优化与监控
  3. ## 5.1 性能指标采集
  4. ```python
  5. import time
  6. import requests
  8. start = time.time()
  9. response = requests.get('https://api.example.com')
  10. latency = time.time() - start
  12. metrics = {
  13.     'status_code': response.status_code,
  14.     'latency_ms': latency * 1000,
  15.     'content_size': len(response.content)
  16. }
  17. print(metrics)

5.2 集成监控系统

对于大规模应用，建议将请求指标接入监控平台：

  1. def monitored_get(url, **kwargs):
  2.     start = time.time()
  3.     try:
  4.         response = requests.get(url, **kwargs)
  5.         latency = time.time() - start
  6.         # 发送指标到监控系统（伪代码）
  7.         send_metric('api_calls', {'url': url, 'status': response.status_code, 'latency': latency})
  8.         return response
  9.     except Exception as e:
  10.         send_metric('api_errors', {'url': url, 'error': str(e)})
  11.         raise

六、总结与扩展建议

Requests库凭借其简洁的API和强大的功能，已成为Python生态中最受欢迎的HTTP客户端。在实际开发中，建议：

  1. 优先使用Session对象管理会话
  2. 实现完善的异常处理和重试机制
  3. 对敏感操作添加日志记录
  4. 定期更新库版本获取安全补丁

对于更复杂的需求，可考虑：

  • 结合aiohttp实现异步请求
  • 使用requests-cache添加本地缓存
  • 通过responses库编写单元测试

通过合理运用这些技术，开发者能够构建出既安全又高效的HTTP客户端应用，满足从简单脚本到企业级服务的各种需求。

最新文章