Преобразование ответа HTTP в словарь JSON

Расширенные примеры работы с JSON из HTTP ответов

1. Потоковая загрузка большого JSON с частичной обработкой

Если JSON очень большой, может потребоваться обрабатывать его по частям, не загружая целиком в память. Используется параметр stream=True в requests и итерация по чанкам с помощью iter_content(), но для JSON это сложно, так как JSON - цельный формат. Альтернатива: использовать ijson для потокового парсинга.

import requests
import ijson

response = requests.get('https://api.example.com/large-data', stream=True)
parser = ijson.parse(response.raw)
for prefix, event, value in parser:
    if event == 'map_key' and value == 'key_of_interest':
        print('Найден ключ:', value)

Пояснение: response.raw предоставляет файловый объект, который передается ijson. Это позволяет обрабатывать JSON, не загружая его целиком.

(вывод зависит от данных)

2. Сохранение JSON ответа в файл и последующее чтение

Иногда ответ нужно сохранить для повторного использования или кэширования.

import requests
import json

resp = requests.get('https://api.example.com/data')
if resp.ok:
    # Сохраняем как есть
    with open('response.json', 'w', encoding='utf-8') as f:
        json.dump(resp.json(), f, ensure_ascii=False, indent=2)
    # Читаем обратно
    with open('response.json', 'r', encoding='utf-8') as f:
        data = json.load(f)
    print(data['key'])

Пояснение: json.dump записывает Python объект в файл. Параметр ensure_ascii=False сохраняет символы Юникода без экранирования.

3. Параллельные запросы с обработкой JSON (concurrent.futures)

Для ускорения сбора данных с нескольких URL используется пул потоков или процессов.

import requests
from concurrent.futures import ThreadPoolExecutor

def fetch_single(url):
    resp = requests.get(url)
    return resp.json() if resp.ok else None

urls = ['https://api1.example.com', 'https://api2.example.com']
with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(fetch_single, urls))
print(results)

Результат: списки словарей для каждого URL.

[{'key': 'value1'}, {'key': 'value2'}]

4. Использование httpx для синхронных и асинхронных запросов

Библиотека httpx поддерживает как синхронный, так и асинхронный интерфейс, а также HTTP/2.

import httpx

# Синхронно
with httpx.Client() as client:
    response = client.get('https://api.example.com/data')
    data = response.json()
    print(data)

# Асинхронно
import asyncio
async def async_fetch():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com/data')
        return response.json()

result = asyncio.run(async_fetch())
print(result)

Пояснение: httpx имеет схожий API с requests, но поддерживает больше возможностей, такие как настраиваемые транспорты и таймауты.

5. Обработка вложенных или сложных структур JSON

Часто JSON содержит вложенные объекты или массивы. Допустим, ответ API содержит список пользователей с адресами.

import requests

resp = requests.get('https://api.example.com/users')
users = resp.json()
# Извлекаем все email
emails = [user['email'] for user in users]
# Группируем по городу
from collections import defaultdict
by_city = defaultdict(list)
for user in users:
    city = user['address']['city']
    by_city[city].append(user['name'])
print(dict(by_city))

Пояснение: типичная работа со списками словарей. Важно проверять наличие ключей, чтобы избежать KeyError.

{
  'New York': ['Alice', 'Bob'],
  'London': ['Charlie']
}

6. Автоматическая обработка пагинации с JSON ответами

Многие API используют пагинацию (например, по страницам). В ответе может быть поле next или аналогичное.

import requests

def fetch_all_pages(base_url, params=None):
    all_data = []
    url = base_url
    while url:
        resp = requests.get(url, params=params)
        data = resp.json()
        all_data.extend(data.get('results', []))
        url = data.get('next')  # url следующей страницы
    return all_data

result = fetch_all_pages('https://api.example.com/items')
print(len(result))

Пояснение: цикл продолжается, пока есть next URL. Нужно учитывать, что API может возвращать next как полный URL или только путь.

7. Работа с JSONP (JSON with Padding)

Редкий случай: сервер возвращает JSONP (функция, оборачивающая JSON). Чтобы получить чистый JSON, нужно извлечь его из строки.

import re
import json

resp_text = 'callback({"key":"value"})'
match = re.search(r'\(\{.*\}\)', resp_text)
if match:
    json_str = match.group()[1:-1]  # убираем скобки
    data = json.loads(json_str)
    print(data)

Пояснение: обычно используется регулярное выражение для выделения объекта JSON внутри скобок.

8. Использование сессий с сохранением cookies и заголовков

Сессия в requests сохраняет cookies между запросами, что полезно для авторизованных API.

session = requests.Session()
session.headers.update({'User-Agent': 'MyApp/1.0'})
session.auth = ('user', 'pass')

resp1 = session.get('https://api.example.com/login')
# cookies сохраняются автоматически
resp2 = session.get('https://api.example.com/secret')
data = resp2.json()
print(data)

9. Обработка ответа с неправильной кодировкой

Иногда сервер не указывает кодировку или указывает неверно. Можно принудительно декодировать в нужной кодировке перед парсингом JSON.

import requests

resp = requests.get('https://api.example.com/data')
raw = resp.content  # байты
try:
    # попытка декодировать как UTF-8
    text = raw.decode('utf-8')
except UnicodeDecodeError:
    # альтернативная кодировка
    text = raw.decode('cp1251')
data = json.loads(text)
print(data)

10. Таймауты и повторные попытки (retry) при получении JSON

Для надежности устанавливают таймаут и автоматически повторяют запрос при временных ошибках.

from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

session = requests.Session()
retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('http://', adapter)
session.mount('https://', adapter)

try:
    resp = session.get('https://api.example.com/data', timeout=5)
    data = resp.json()
except requests.exceptions.RequestException as e:
    print('Не удалось получить данные:', e)

Пояснение: настройка retry автоматически повторяет запрос при указанных ошибках сервера.