Python实现简单封装网络请求的示例详解
作者：冰冷的bin
这篇文章主要为大家详细介绍了Python实现简单封装网络请求的相关知识,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下
安装依赖
在开始使用前，需先安装工具依赖的第三方库。打开终端（命令行），执行以下命令：
pip install requests tenacity
核心功能说明
1. 类与方法概览
| 类/方法 | 功能描述 |
|---|---|
| ApiResponse 类 | 响应数据包装类，提供状态检查、数据获取等便捷方法 |
| 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",
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,
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,
timeout=self.timeout,
verify=self.verify_ssl
)
# 解析响应
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封装网络请求内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！