Python响应解析案例如何解析返回值

wen python案例 2026-06-30 31

本文目录导读：

Python响应解析案例如何解析返回值

使用requests库解析响应
处理嵌套JSON响应
处理不同HTTP状态码
异步响应解析
实用工具函数

我来为您详细介绍Python中解析HTTP响应返回值的各种情况和方法。

使用requests库解析响应

基本JSON响应解析

import requests
# 发送请求
response = requests.get('https://api.example.com/users')
# 解析JSON响应
if response.status_code == 200:
    data = response.json()  # 自动解析JSON
    print(f"状态码: {response.status_code}")
    print(f"响应头: {response.headers}")
    print(f"解析后的数据: {data}")
    # 访问JSON数据
    if isinstance(data, dict):
        print(f"用户列表: {data.get('users', [])}")
    elif isinstance(data, list):
        print(f"用户数: {len(data)}")

处理不同类型的响应

import requests
import json
def parse_response(response):
    """智能解析各种类型的响应"""
    # 1. 获取原始内容
    content = response.content  # 字节格式
    text = response.text       # 字符串格式
    # 2. 根据Content-Type解析
    content_type = response.headers.get('content-type', '').lower()
    if 'application/json' in content_type:
        # JSON格式
        try:
            return response.json()
        except json.JSONDecodeError as e:
            print(f"JSON解析失败: {e}")
            return None
    elif 'text/xml' in content_type or 'application/xml' in content_type:
        # XML格式
        import xml.etree.ElementTree as ET
        try:
            root = ET.fromstring(response.text)
            return root
        except ET.ParseError as e:
            print(f"XML解析失败: {e}")
            return None
    elif 'text/html' in content_type:
        # HTML格式
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(response.text, 'html.parser')
        return soup
    elif 'application/x-www-form-urlencoded' in content_type:
        # URL编码格式
        from urllib.parse import parse_qs
        return parse_qs(response.text)
    else:
        # 文本或其他格式
        return response.text
# 使用示例
response = requests.get('https://api.example.com/data')
parsed_data = parse_response(response)
print(f"解析结果类型: {type(parsed_data)}")

处理嵌套JSON响应

import requests
import json
from typing import Dict, List, Any
class ResponseParser:
    """响应解析器"""
    def __init__(self, response: requests.Response):
        self.response = response
        self.raw_data = None
    def parse_json(self) -> Dict[str, Any]:
        """解析标准JSON响应"""
        try:
            self.raw_data = self.response.json()
            return self.raw_data
        except Exception as e:
            print(f"解析错误: {e}")
            return {}
    def extract_value(self, key_path: str, default=None):
        """
        提取嵌套JSON中的值
        key_path: 用点号分隔的键路径，如 'data.user.name'
        """
        if not self.raw_data:
            self.parse_json()
        keys = key_path.split('.')
        current = self.raw_data
        for key in keys:
            if isinstance(current, dict):
                current = current.get(key)
                if current is None:
                    return default
            else:
                return default
        return current
    def get_paginated_data(self, data_path: str, page_param: str = 'page'):
        """
        处理分页数据
        """
        all_data = []
        page = 1
        while True:
            # 请求当前页
            params = {page_param: page}
            resp = requests.get(self.response.url, params=params)
            if resp.status_code != 200:
                break
            # 提取当前页数据
            data = resp.json()
            items = self._nested_get(data, data_path)
            if not items:
                break
            all_data.extend(items)
            # 检查是否有下一页
            if not self._has_next_page(data):
                break
            page += 1
        return all_data
    def _nested_get(self, data: dict, path: str):
        """内部方法：获取嵌套数据"""
        keys = path.split('.')
        current = data
        for key in keys:
            if isinstance(current, dict):
                current = current.get(key)
            else:
                return None
        return current
    def _has_next_page(self, data: dict) -> bool:
        """检查是否有下一页"""
        # 根据不同API的实现调整
        return data.get('has_next', False) or data.get('next_page') is not None
# 使用示例
response = requests.get('https://api.example.com/users/1')
parser = ResponseParser(response)
# 解析完整数据
data = parser.parse_json()
print(f"完整数据: {json.dumps(data, indent=2, ensure_ascii=False)}")
# 提取特定字段
name = parser.extract_value('data.user.name', '未知用户')
email = parser.extract_value('data.user.contact.email')
print(f"用户名: {name}, 邮箱: {email}")

处理不同HTTP状态码

import requests
def handle_response(response: requests.Response):
    """根据状态码处理响应"""
    status_code = response.status_code
    if status_code == 200:
        # 成功响应
        try:
            return response.json()
        except:
            return response.text
    elif status_code == 201:
        # 创建成功
        print("资源创建成功")
        return response.json()
    elif status_code == 204:
        # 无内容
        print("操作成功，无返回内容")
        return None
    elif status_code == 301 or status_code == 302:
        # 重定向
        print(f"重定向到: {response.headers.get('Location')}")
        # 可以跟随重定向
        follow_response = requests.get(response.headers['Location'])
        return handle_response(follow_response)
    elif status_code == 400:
        # 错误的请求
        error_data = response.json()
        print(f"请求错误: {error_data.get('message', '未知错误')}")
        return {'error': error_data}
    elif status_code == 401:
        # 未授权
        print("需要认证，请检查API密钥")
        return {'error': 'Unauthorized'}
    elif status_code == 403:
        # 禁止访问
        print("访问被拒绝")
        return {'error': 'Forbidden'}
    elif status_code == 404:
        # 资源不存在
        print("请求的资源不存在")
        return {'error': 'Not Found'}
    elif status_code == 429:
        # 请求过多
        retry_after = response.headers.get('Retry-After', '60')
        print(f"请求速度过快，请在{retry_after}秒后重试")
        return {'error': 'Rate Limited', 'retry_after': retry_after}
    elif 500 <= status_code < 600:
        # 服务器错误
        print(f"服务器错误 ({status_code})")
        return {'error': f'Server Error: {status_code}'}
    else:
        # 其他状态码
        print(f"未处理的HTTP状态码: {status_code}")
        return {'error': f'Unhandled status code: {status_code}'}
# 使用示例
urls = [
    'https://api.example.com/success',
    'https://api.example.com/not-found',
    'https://api.example.com/error'
]
for url in urls:
    response = requests.get(url)
    result = handle_response(response)
    print(f"URL: {url}")
    print(f"结果: {result}\n")

异步响应解析

import aiohttp
import asyncio
import json
async def parse_async_response(session: aiohttp.ClientSession, url: str):
    """异步解析API响应"""
    try:
        async with session.get(url) as response:
            # 获取状态码
            status = response.status
            # 获取响应头
            headers = dict(response.headers)
            # 异步获取文本内容
            text = await response.text()
            # 尝试解析JSON
            try:
                data = json.loads(text)
                return {
                    'status': status,
                    'headers': headers,
                    'data': data,
                    'type': 'json'
                }
            except json.JSONDecodeError:
                # 如果不是JSON，返回文本
                return {
                    'status': status,
                    'headers': headers,
                    'data': text,
                    'type': 'text'
                }
    except aiohttp.ClientError as e:
        print(f"请求错误: {e}")
        return {'status': 0, 'error': str(e)}
async def parse_multiple_responses(urls: list):
    """并发解析多个API响应"""
    async with aiohttp.ClientSession() as session:
        tasks = [parse_async_response(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results
# 使用示例
async def main():
    urls = [
        'https://api.example.com/endpoint1',
        'https://api.example.com/endpoint2'
    ]
    results = await parse_multiple_responses(urls)
    for i, result in enumerate(results):
        print(f"结果 {i+1}:")
        print(f"  状态: {result.get('status')}")
        print(f"  类型: {result.get('type')}")
        print(f"  数据: {result.get('data')}\n")
# 运行异步函数
# asyncio.run(main())

实用工具函数

import requests
import json
from datetime import datetime
def parse_api_response(response: requests.Response) -> dict:
    """通用的API响应解析函数"""
    result = {
        'success': False,
        'status_code': response.status_code,
        'data': None,
        'error': None,
        'meta': {}
    }
    try:
        # 检查状态码
        if response.ok:
            # 尝试解析JSON
            try:
                result['data'] = response.json()
            except ValueError:
                result['data'] = response.text
            result['success'] = True
            # 解析响应头中的元信息
            result['meta'] = {
                'content_type': response.headers.get('content-type'),
                'content_length': response.headers.get('content-length'),
                'server': response.headers.get('server'),
                'date': response.headers.get('date'),
                'request_time': response.elapsed.total_seconds()
            }
        else:
            # 错误响应
            try:
                error_data = response.json()
                result['error'] = error_data.get('message', error_data.get('error', '未知错误'))
            except ValueError:
                result['error'] = response.text[:200]  # 只取前200个字符
    except requests.RequestException as e:
        result['error'] = str(e)
    return result
# 使用示例
def main():
    # 模拟不同的API响应
    responses = [
        requests.Response(),  # 空响应示例
    ]
    api_response = parse_api_response(responses[0] if responses else requests.Response())
    if api_response['success']:
        print(f"请求成功!")
        print(f"数据: {json.dumps(api_response['data'], indent=2)}")
        print(f"元信息: {api_response['meta']}")
    else:
        print(f"请求失败: {api_response['error']}")
if __name__ == "__main__":
    main()

这些示例涵盖了Python中解析HTTP响应的主要场景,包括：

JSON解析：最常用的API响应格式
错误处理：各种HTTP状态码的适当处理
嵌套数据提取：从复杂JSON结构中提取值
分页处理：处理分页API响应
异步解析：提高大量请求的性能
通用工具函数：可复用的解析逻辑

根据您的具体需求选择合适的解析方法。