Python实现简单封装网络请求的示例详解

目录

• 安装依赖
• 核心功能说明
• 1. 类与方法概览
• 2.NetHelper类初始化参数
• 3.ApiResponse类属性与方法
• 使用实例
• 步骤 1：导入工具类
• 步骤 2：初始化NetHelper实例
• 步骤 3：发送登录请求
• 步骤 4：处理响应结果
• 五、源码

安装依赖

在开始使用前，需先安装工具依赖的第三方库。打开终端（命令行），执行以下命令：

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",
    编程客栈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,
                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封装网络请求内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！