如何在Flutter中使用Riverpod创建通用API类以获取统一响应

核心思路 #

封装通用HTTP请求方法，强制统一响应输出格式，同时预留扩展点适配不同API的参数、请求头及响应结构差异。

代码实现（Python） #

import requests
from typing import Dict, Optional, Any

class UniversalAPIClient:
    def __init__(self, base_url: str, default_headers: Optional[Dict[str, str]] = None):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update(default_headers or {})

    def _process_response(self, response: requests.Response) -> Dict[str, Any]:
        """统一处理响应，转换成固定格式"""
        try:
            response.raise_for_status()  # 抛出HTTP错误
            response_data = response.json()
        except requests.exceptions.HTTPError as e:
            return {
                "status": "error",
                "data": None,
                "message": f"HTTP错误: {str(e)}",
                "status_code": response.status_code
            }
        except ValueError:
            return {
                "status": "error",
                "data": response.text,
                "message": "响应不是JSON格式",
                "status_code": response.status_code
            }
        except requests.exceptions.RequestException as e:
            return {
                "status": "error",
                "data": None,
                "message": f"请求异常: {str(e)}",
                "status_code": None
            }

        # 统一成功响应格式
        return {
            "status": "success",
            "data": response_data,
            "message": "请求成功",
            "status_code": response.status_code
        }

    def get(self, endpoint: str, params: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.get(url, params=params, headers=headers)
        return self._process_response(response)

    def post(self, endpoint: str, data: Optional[Dict[str, Any]] = None, json: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.post(url, data=data, json=json, headers=headers)
        return self._process_response(response)

    # 可扩展put、delete等其他HTTP方法

使用示例 #

# 初始化客户端，设置默认请求头
client = UniversalAPIClient("https://api.example.com", default_headers={"Authorization": "Bearer YOUR_TOKEN"})

# 调用GET接口
get_response = client.get("users", params={"page": 1})
print(get_response)

# 调用POST接口
post_response = client.post("users", json={"name": "John", "email": "john@example.com"})
print(post_response)

统一响应的关键要点 #

• 固定响应结构：无论调用哪个API，都返回包含status（success/error）、data（响应数据）、message（提示信息）、status_code（HTTP状态码）的字典，便于上层逻辑统一处理。
• 异常捕获全覆盖：处理网络异常、HTTP错误、非JSON响应等多种情况，避免因API返回异常导致上层代码崩溃。
• 灵活扩展：通过default_headers和请求方法的headers参数适配不同API的认证需求；支持params、data、json等不同参数传递方式。
• 会话复用：使用requests.Session复用TCP连接，提升请求效率。

内容的提问来源于stack exchange，提问作者Umesh Rajput