Python中如何解析JSON数据 处理API响应时有哪些技巧

import json
import requests

def safe_json_parse(response):
    try:
        if response.headers['Content-Type'] == 'application/json':
            return response.json()
        else:
            return {'error': 'Not a JSON response', 'content': response.text}
    except json.JSONDecodeError as e:
        return {'error': 'JSONDecodeError', 'details': str(e)}
    except KeyError: # 处理Content-Type不存在的情况
        return {'error': 'Content-Type not found', 'content': response.text}
    except Exception as e:
        return {'error': 'Unexpected error', 'details': str(e)}

response = requests.get('https://api.example.com/data') # 假设这个API可能返回非JSON
data = safe_json_parse(response)

if 'error' in data:
    print(f"Error: {data['error']}")
    if 'details' in data:
        print(f"Details: {data['details']}")
else:
    # 正常处理数据
    print(data)

import ijson
import requests

def process_large_json(url):
    response = requests.get(url, stream=True) # 启用流式传输
    objects = ijson.items(response.raw, 'item') # 假设JSON是一个包含多个item的数组
    for item in objects:
        # 处理每个item，例如保存到数据库
        print(item['id'], item['name']) # 假设每个item都有id和name字段
        # ...

process_large_json('https://api.example.com/large_data.json')

import requests

def fetch_all_data(base_url, page_param='page', page_size_param='page_size', page_size=100):
    all_data = []
    page = 1
    while True:
        url = f"{base_url}?{page_param}={page}&{page_size_param}={page_size}"
        response = requests.get(url)
        response.raise_for_status() # 检查HTTP错误
        data = response.json()
        if not data: # 假设空列表表示没有更多数据
            break
        all_data.extend(data)
        page += 1
    return all_data

all_data = fetch_all_data('https://api.example.com/items')
print(f"Total items fetched: {len(all_data)}")

import jsonschema
import requests

schema = {
    "type": "array",
    "items": {
        "type": "object",
        "properties": {
            "id": {"type": "integer"},
            "name": {"type": "string"},
            "price": {"type": "number"}
        },
        "required": ["id", "name", "price"]
    }
}

def validate_data(data, schema):
    try:
        jsonschema.validate(instance=data, schema=schema)
        return True
    except jsonschema.exceptions.ValidationError as e:
        print(f"Validation error: {e}")
        return False

response = requests.get('https://api.example.com/products')
data = response.json()

if validate_data(data, schema):
    print("Data is valid")
    # 处理数据
else:
    print("Data is invalid")

import requests
import time

def handle_rate_limit(url, max_retries=5, wait_time=60):
    for attempt in range(max_retries):
        response = requests.get(url)
        if response.status_code == 429:
            print(f"Rate limit exceeded. Waiting {wait_time} seconds before retry (attempt {attempt + 1}/{max_retries})")
            time.sleep(wait_time)
        else:
            response.raise_for_status() # 检查其他HTTP错误
            return response.json()
    raise Exception("Max retries exceeded")

data = handle_rate_limit('https://api.example.com/limited_resource')
print(data)

import requests_cache

session = requests_cache.CachedSession('my_cache') # 创建一个缓存会话

response = session.get('https://api.example.com/cached_data')
data = response.json()

print(response.from_cache) # 检查数据是否来自缓存
print(data)

import asyncio
import aiohttp

async def fetch_data(session, url):
    async with session.get(url) as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        urls = ['https://api.example.com/data1', 'https://api.example.com/data2', 'https://api.example.com/data3']
        tasks = [fetch_data(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(result)

asyncio.run(main())