Python中如何解析JSON數(shù)據(jù) 處理API響應(yīng)時(shí)有哪些技巧

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') # 假設(shè)這個(gè)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:     # 正常處理數(shù)據(jù)     print(data)

import ijson import requests  def process_large_json(url):     response = requests.get(url, stream=True) # 啟用流式傳輸     objects = ijson.items(response.raw, 'item') # 假設(shè)JSON是一個(gè)包含多個(gè)item的數(shù)組     for item in objects:         # 處理每個(gè)item，例如保存到數(shù)據(jù)庫(kù)         print(item['id'], item['name']) # 假設(shè)每個(gè)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錯(cuò)誤         data = response.json()         if not data: # 假設(shè)空列表表示沒(méi)有更多數(shù)據(jù)             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")     # 處理數(shù)據(jù) 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錯(cuò)誤             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') # 創(chuàng)建一個(gè)緩存會(huì)話(huà)  response = session.get('https://api.example.com/cached_data') data = response.json()  print(response.from_cache) # 檢查數(shù)據(jù)是否來(lái)自緩存 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())