Python接口工具案例如何封装接口请求

本文目录导读：

1. 基础封装版（最常用）
2. 进阶封装版（带日志、重试、自定义异常）
3. 带Token认证的封装
4. 项目使用目录结构建议
5. 最佳实践总结

下面给你一个Python接口请求封装的完整案例，包含常见的封装模式、异常处理、日志、重试机制等，适合实际项目使用。

基础封装版（最常用）

使用 requests 库，封装 GET、POST、PUT、DELETE 等常见方法。

import requests
import json
class ApiClient:
    """接口请求封装类"""
    def __init__(self, base_url="", headers=None):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.headers = headers or {
            "Content-Type": "application/json",
            "Accept": "application/json"
        }
        self.session.headers.update(self.headers)
    def _request(self, method, endpoint, **kwargs):
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        try:
            response = self.session.request(method, url, **kwargs)
            response.raise_for_status()  # 非200状态会抛出异常
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {e}")
            return None
    def get(self, endpoint, params=None):
        return self._request("GET", endpoint, params=params)
    def post(self, endpoint, data=None, json_data=None):
        return self._request("POST", endpoint, data=data, json=json_data)
    def put(self, endpoint, data=None, json_data=None):
        return self._request("PUT", endpoint, data=data, json=json_data)
    def delete(self, endpoint):
        return self._request("DELETE", endpoint)
# 使用示例
client = ApiClient(base_url="https://api.example.com")
result = client.get("/users", params={"page": 1})
print(result)

进阶封装版（带日志、重试、自定义异常）

import requests
import logging
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# 配置日志
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class APIClientError(Exception):
    """自定义API异常"""
    pass
class AdvancedApiClient:
    """进阶接口请求封装"""
    def __init__(self, base_url="", headers=None, timeout=10, max_retries=3):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        # 设置重试策略
        retry_strategy = Retry(
            total=max_retries,
            backoff_factor=0.5,  # 重试间隔倍数
            status_forcelist=[500, 502, 503, 504],  # 哪些状态码需要重试
            allowed_methods=["GET", "POST", "PUT", "DELETE"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
        # 设置默认请求头
        self.headers = headers or {
            "Content-Type": "application/json",
            "Accept": "application/json"
        }
        self.session.headers.update(self.headers)
    def _request(self, method, endpoint, **kwargs):
        """统一的请求处理方法"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        kwargs.setdefault("timeout", self.timeout)
        try:
            logger.info(f"发送 {method} 请求: {url}")
            response = self.session.request(method, url, **kwargs)
            # 打印响应信息
            logger.info(f"响应状态码: {response.status_code}")
            # 检查HTTP状态码
            response.raise_for_status()
            # 尝试解析JSON
            try:
                return response.json()
            except (json.JSONDecodeError, ValueError):
                logger.warning("响应不是JSON格式，返回原始文本")
                return response.text
        except requests.exceptions.Timeout:
            logger.error(f"请求超时: {url}")
            raise APIClientError(f"请求超时: {url}")
        except requests.exceptions.HTTPError as e:
            status_code = e.response.status_code if e.response else "未知"
            logger.error(f"HTTP错误 {status_code}: {url}")
            raise APIClientError(f"HTTP错误 {status_code}: {e}")
        except requests.exceptions.ConnectionError:
            logger.error(f"连接失败: {url}")
            raise APIClientError(f"连接失败: {url}")
        except Exception as e:
            logger.error(f"未知错误: {e}")
            raise APIClientError(f"请求失败: {e}")
    def get(self, endpoint, params=None, **kwargs):
        return self._request("GET", endpoint, params=params, **kwargs)
    def post(self, endpoint, data=None, json=None, **kwargs):
        return self._request("POST", endpoint, data=data, json=json, **kwargs)
    def put(self, endpoint, data=None, json=None, **kwargs):
        return self._request("PUT", endpoint, data=data, json=json, **kwargs)
    def delete(self, endpoint, **kwargs):
        return self._request("DELETE", endpoint, **kwargs)
# 使用示例
try:
    client = AdvancedApiClient(base_url="https://api.example.com", max_retries=3)
    data = {"name": "John", "age": 30}
    result = client.post("/users", json=data)
    print("创建用户成功:", result)
except APIClientError as e:
    print(f"API调用失败: {e}")

带Token认证的封装

class AuthApiClient(AdvancedApiClient):
    """带Token认证的API客户端"""
    def __init__(self, base_url, token=None, **kwargs):
        super().__init__(base_url, **kwargs)
        if token:
            self.set_token(token)
    def set_token(self, token):
        """设置认证Token"""
        self.session.headers.update({
            "Authorization": f"Bearer {token}"
        })
    def login(self, username, password, login_endpoint="/auth/login"):
        """登录并保存Token"""
        data = {"username": username, "password": password}
        result = self.post(login_endpoint, json=data)
        if result and "token" in result:
            self.set_token(result["token"])
            logger.info("登录成功，Token已保存")
            return result
        raise APIClientError("登录失败，未获取到Token")

项目使用目录结构建议

api_client/
├── __init__.py
├── base.py          # 基础封装
├── auth.py          # 认证相关
├── exceptions.py    # 自定义异常
├── config.py        # 配置（URL、超时等）
└── utils.py         # 工具函数（如数据清洗）

最佳实践总结

如果你有具体的接口（如RESTful API、GraphQL、WebSocket等）或特定的需求（如文件上传、批量请求），可以告诉我，我可以帮你定制封装方案。