本文主要是介绍Python实现简单封装网络请求的示例详解，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
《Python实现简单封装网络请求的示例详解》这篇文章主要为大家详细介绍了Python实现简单封装网络请求的相关知识,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下...
安装依赖
在开始使用前，需先安装工具依赖的第三方库。打开终端（命令行），执行以下命令：
pip install requests tenacity
核心功能说明
1. 类与方法概览
| 类/方法 | 功能描述 |
|---|---|
| ApiResponse 类 | 响应数据包装类，提供状态js检查、数据获取等便捷方法 |
| NetHelper 类 | 核心请求封装类，支持初始化配置、多 HTTP 方法调用 |
| get()/post()/put()/delete() | 对外暴露的快捷方法，分别对应 GET/POST/PUT/DELETE 请求 |
2.NetHelper类初始化参数
初始化 NetHelper 实例时，需传入以下参数（均为可选，有默认值）：
| 参数名 | 类型 | 说明 |
|---|---|---|
| base_url | str | 基础 URL（如 https://api.example.com），请求路径会自动拼接至此 URL |
| headers | Dict[str, str] | 默认请求头（会被单次请求的头覆盖） |
| timeout | int | 请求超时时间（秒，默认 15 秒） |
| verify_ssl | bool | 是否验证 SSL 证书（生产环境建议设为 True，测试环境可设为 False） |
3.ApiResponse类属性与方法
ApiResponse 用于包装接口响应数据，提供以下核心功能：
| 属性/方法 | 类型 | 说明 |
|---|---|---|
| data | Any | 响应体解析后的数据（jsON 自动转为字典，非 JSON 为文本/二进制） |
| status_code | int | HTTP 状态码（如 200、404） |
| error | Optional[str] | 错误信息（仅当状态码非 2xx 或发生异常时存在） |
| is_success() | bool | 检查请求是否成功（状态码 2xx 且无错误信息时返回 True） |
使用实例
步骤 1：导入工具类
from net_helper import NetHelper, ApiResponse # 假设工具保存为 net_helper.py
步骤 2：初始化NetHelper实例
请求头和基础 URL 配置：
# 初始化 NetHelper
client = NetHelper(
base_url="https://com.text.com",
headers={
"Accept": "application/json, text/plain, */*",
"Content-Type": "application/json;charset=UTF-8",
},
)
步骤 3：发送登录请求
使用 post() 方法发送 POST 请求，传入请求体和参数：
response = client.post(
url="/login",
json={
"name": "bin",
"pwd": "123456",
}
)
步骤 4：处理响应结果
根据 ApiResponse 的属性判断请求是否成功，并获取数据：
if response.is_success():
print("登录成功！")
print("响应数据:", response.data)
print("状态码:", response.status_code)
else:
print(f"登录失败！错误信息: {response.error}")
print("状态码:", response.status_code)
print("原始响应:", response.data)
五、源码
#!/usr/bin/env python3 # -*- coding: utf-8 -*- import logging import json as jsonlib from typing import Dict, Optional, Union, Any import requests from tenacity import ( retry, stop_after_attempt, wait_exponential, retry_if_exception_type, ) logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", China编程datefmt="%Y-%m-%d %H:%M:%S" ) logger = logging.getLogger(__name__) class ApiResponse: """API响应包装类""" def __init__(self, data: Any, status_code: int = 200, error: Optional[str] = None): self.data = data self.status_code = status_code self.error = error self.message = data.get('message', '') if isinstance(data, dict) else '' def is_success(self) -> bool: """检查请求是否成功""" return 200 <= self.status_code < 300 and self.error is None class NetHelper: def __init__(self, base_url: str, headers: Optional[Dict[str, str]], timeout: int = 15, verify_ssl: bool = True): self.base_url = base_url.rstrip("/") self.default_headers = headers self.timeout = timeout self.verify_ssl = verify_ssl @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), retry=retry_if_exception_type(( requests.exceptions.ConnectionError, requests.exceptions.Timeout )), reraise=True ) def _send_request( self, method: str = "GET", url: str = "", headers: Optional[Dict[str, str]] = None, 编程China编程 json: Optional[Union[Dict[str, Any], str]] = None, data: Optional[Union[Dict[str, Any], str, bytes]] = None, params: Optional[Dict[str, Any]] = None ) -> ApiResponse: """内部请求方法""" # 拼接完整URL full_url = f"{self.base_url}/{url.lstrip('/')}" # 合并请求头 merged_headers = {**self.default_headers, **(headers or {})} # 移除空值头 merged_headers = {k: v for k, v in merged_headers.items() if v is not None} try: # 打印请求体 print("请求方法:", method.upper()) print("请求URL:", full_url) if params: print("请求参数:") print(jsonlib.dumps(params, ensure_ascii=False, indent=2)) if json: print("请求JSON数据:") print(jsonlib.dumps(json, ensure_ascii=False, indent=2)) if data: print("请求表单数据:") if isinstance(data, (dict, list)): print(jsonlib.dumps(data, ensure_ascii=False, indent=2)) else: print(data) # 发送请求 response = requests.request( method=method.upper(), url=full_url, headers=merged_headers, json=json, data=data, params=params, timeoutphp=self.timeout, verify=self.verify_ssl javascript ) # 解析响应 try: result_data = response.json() except ValueError: result_data = response.text if response.encoding else response.content print("响应数据:") print(jsonlib.dumps(result_data, ensure_ascii=False, indent=2)) # 检查HTTP状态码 if 200 <= response.status_code < 300: return ApiResponse(result_data, response.status_code) else: error_msg = f"HTTP错误: 状态码 {response.status_code}" logger.error(f"{error_msg}, 响应: {result_data}") return ApiResponse(result_data, response.status_code, error_msg) except requests.exceptions.ConnectionError as e: logger.error(f"连接失败: 无法连接到 {full_url}") return ApiResponse({}, 0, str(e)) except requests.exceptions.Timeout as e: logger.error(f"请求超时: 超过 {self.timeout} 秒") return ApiResponse({}, 0, str(e)) except Exception as e: logger.error(f"未知错误: {str(e)}") return ApiResponse({}, 0, str(e)) def post(self, url: str, data: Optional[Any] = None, params: Optional[Dict[str, Any]] = None) -> ApiResponse: return self._send_request(method="POST", url=url, json=data, params=params) def get(self, url: str, params: Optional[Dict[str, Any]] = None) -> ApiResponse: return self._send_request(method="GET", url=url, params=params) def put(self, url: str, data: Optional[Any] = None, params: Optional[Dict[str, Any]] = None) -> ApiResponse: return self._send_request(method="PUT", url=url, json=data, params=params) def delete(self, url: str, params: Optional[Dict[str, Any]] = None) -> ApiResponse: return self._send_request(method="DELETE", url=url, params=params)
到此这篇关于Python实现简单封装网络请求的示例详解的文章就介绍到这了,更多相关Python封装网络请求内容请搜索China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持China编程(www.chinasem.cn)！
这篇关于Python实现简单封装网络请求的示例详解的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
